feat

build.ipynb

@@ -426,28 +426,6 @@
     "!echo \"kubectl create secret docker-registry regcred --docker-server=$REGISTRY_HOST --docker-username='user' --docker-password=$REGISTRY_PASS --docker-email=\\\"$(git config --global --get user.email)\\\" && clear && history -c && history -w\""
    ]
   },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "id": "5aed91fa",
-   "metadata": {
-    "vscode": {
-     "languageId": "shellscript"
-    }
-   },
-   "outputs": [],
-   "source": [
-    "%%bash\n",
-    "set -euo pipefail\n",
-    "EXPANDED=\"$(envsubst < k8s/tg.yaml)\"\n",
-    "echo \"$EXPANDED\"\n",
-    "echo \"kubectl apply -f - <<'EOF'\"\n",
-    "echo \"$(envsubst < k8s/tg.yaml)\"\n",
-    "echo \"EOF\"\n",
-    "echo \"clear\"\n",
-    "echo \"history -c\"\n"
-   ]
-  },
   {
    "cell_type": "code",
    "execution_count": null,

k8s/tg.yaml

@@ -29,6 +29,8 @@ metadata:
 type: Opaque
 stringData:
   RABBITMQ_PASS: $RABBITMQ_PASS
+  RABBITMQ_HOST: $RABBITMQ_HOST
+  RABBITMQ_PORT: '$RABBITMQ_PORT'
   BOT_TOKEN: "$BOT_TOKEN"
   ATOKEN: "$ATOKEN"
   FULL_PATH_DOMAIN: $FULL_PATH_DOMAIN

main.ipynb

@@ -1908,10 +1908,14 @@
     "    data: PartnerExhibitionDelete = Body(...),\n",
     "    current_user: Dict = Security(get_current_user, scopes=[\"create:super_items\"])\n",
     "):\n",
-    "    await exhibition_collection.delete_one(\n",
+    "    exhibition = await exhibition_collection.find_one_and_delete(\n",
     "        {'_id':ObjectId(data.exhibition_id), 'user._id': current_user.get('_id')},\n",
     "    )\n",
-    "    return {'ok': True}"
+    "    for i in exhibition.get('tg',[]):\n",
+    "        tg.delete_from_tg.delay(i.get('result',[{}])[0].get('message_id'), i.get('doc'))\n",
+    "    for i in exhibition.get('vk',[]):\n",
+    "        vk.delete_wall.delay(i.get('response',{}).get('post_id'), i.get('doc'))\n",
+    "    return {'ok': True}\n"
    ]
   },
   {
@@ -1926,10 +1930,14 @@
     "    data: Partnercompetitionstop = Body(...),\n",
     "    current_user: Dict = Security(get_current_user, scopes=[\"create:super_items\"])\n",
     "):\n",
-    "    await competitions_col.delete_one(\n",
+    "    competition = await competitions_col.delete_one(\n",
     "        {'_id':ObjectId(data.competition_id), 'user._id': current_user.get('_id')},\n",
     "    )\n",
-    "    return {'ok': True}"
+    "    for i in competition.get('tg',[]):\n",
+    "        tg.delete_from_tg.delay(i.get('result',[{}])[0].get('message_id'), i.get('doc'))\n",
+    "    for i in competition.get('vk',[]):\n",
+    "        vk.delete_wall.delay(i.get('response',{}).get('post_id'), i.get('doc'))\n",
+    "    return {'ok': True}\n"
    ]
   },
   {

main.py

@@ -1450,21 +1450,31 @@ async def partner_exhibition_delete(
     data: PartnerExhibitionDelete = Body(...),
     current_user: Dict = Security(get_current_user, scopes=["create:super_items"])
 ):
-    await exhibition_collection.delete_one(
+    exhibition = await exhibition_collection.find_one_and_delete(
         {'_id':ObjectId(data.exhibition_id), 'user._id': current_user.get('_id')},
     )
+    for i in exhibition.get('tg',[]):
+        tg.delete_from_tg.delay(i.get('result',[{}])[0].get('message_id'), i.get('doc'))
+    for i in exhibition.get('vk',[]):
+        vk.delete_wall.delay(i.get('response',{}).get('post_id'), i.get('doc'))
     return {'ok': True}
 
+
 @app.delete("/api/partner/competition/delete")
 async def partner_competition_delete(
     data: Partnercompetitionstop = Body(...),
     current_user: Dict = Security(get_current_user, scopes=["create:super_items"])
 ):
-    await competitions_col.delete_one(
+    competition = await competitions_col.delete_one(
         {'_id':ObjectId(data.competition_id), 'user._id': current_user.get('_id')},
     )
+    for i in competition.get('tg',[]):
+        tg.delete_from_tg.delay(i.get('result',[{}])[0].get('message_id'), i.get('doc'))
+    for i in competition.get('vk',[]):
+        vk.delete_wall.delay(i.get('response',{}).get('post_id'), i.get('doc'))
     return {'ok': True}
 
+
 @app.post("/api/partner/competition/finish")
 async def partner_competition_finish(
     data: Partnercompetitionstop = Body(...),

tg.ipynb

@@ -322,6 +322,56 @@
     "    raise last_exc\n"
    ]
   },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "fc96bf02",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "def delete_message(chat_id: int, message_id: int) -> bool:\n",
+    "    API_URL = f\"https://api.telegram.org/bot{BOT_TOKEN}/deleteMessage\"\n",
+    "    payload = {\n",
+    "        \"chat_id\": chat_id,\n",
+    "        \"message_id\": message_id,\n",
+    "    }\n",
+    "    # Telegram expects a POST request with JSON body\n",
+    "    with httpx.Client() as client:\n",
+    "        response = client.post(API_URL, json=payload, timeout=10)\n",
+    "    # The API returns JSON like {\"ok\":true,\"result\":true}\n",
+    "    data = response.json()\n",
+    "    return data.get(\"ok\") and data.get(\"result\") is True"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "cc5733db",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "@celery_app_tg.task(\n",
+    "    bind=True,\n",
+    "    rate_limit='60/m',\n",
+    "    name='tg.delete_message',\n",
+    "    retry_backoff=True,\n",
+    "    retry_backoff_max=600,\n",
+    "    retry_jitter=True,\n",
+    "    max_retries=20,\n",
+    "    soft_time_limit=600,\n",
+    "    time_limit=700,\n",
+    ")\n",
+    "def delete_from_tg(self, message_id: str, social: dict):\n",
+    "    try:\n",
+    "        chat_id = social.get('screen_name')\n",
+    "        logging.info(f\"TG удаление публикации {chat_id} {message_id}\")\n",
+    "        result = asyncio.run(delete_message(chat_id, message_id))\n",
+    "        # result['doc'] = social\n",
+    "        # httpx.post(f'{FULL_PATH_DOMAIN}/tg', json={'item': item, 'result': result}, headers={\"Authorization\": f\"bearer {ATOKEN}\"})\n",
+    "    except Exception as exc:\n",
+    "        raise self.retry(exc=exc, countdown=30)\n"
+   ]
+  },
   {
    "cell_type": "code",
    "execution_count": null,
@@ -429,16 +479,6 @@
     "## Test"
    ]
   },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "id": "7b713b9a",
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "https://t.me/kultmost/2594"
-   ]
-  },
   {
    "cell_type": "code",
    "execution_count": null,
@@ -446,52 +486,7 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "a=\"\"\"2581\n",
+    "a=\"\"\"2645\"\"\".split('\\n')\n",
-    "2594\n",
-    "2582\n",
-    "2584\n",
-    "2586\n",
-    "2585\n",
-    "2589\n",
-    "2588\n",
-    "2590\n",
-    "2591\n",
-    "2592\n",
-    "2593\n",
-    "2596\n",
-    "2595\n",
-    "2599\n",
-    "2597\n",
-    "2604\n",
-    "2602\n",
-    "2598\n",
-    "2603\n",
-    "2606\n",
-    "2600\n",
-    "2607\n",
-    "2608\n",
-    "2601\n",
-    "2612\n",
-    "2605\n",
-    "2610\n",
-    "2616\n",
-    "2609\n",
-    "2626\n",
-    "2615\n",
-    "2611\n",
-    "2627\n",
-    "2628\n",
-    "2614\n",
-    "2613\n",
-    "2618\n",
-    "2623\n",
-    "2625\n",
-    "2624\n",
-    "2619\n",
-    "2617\n",
-    "2621\n",
-    "2620\n",
-    "2622\"\"\".split('\\n')\n",
     "posts = map(int,a)\n",
     "import httpx\n",
     "import os\n",

tg.py

@@ -207,6 +207,41 @@ async def send_tg(urls: list, text: str, chat_id: str, sign_avatar: bool = True,
     raise last_exc
 
 
+def delete_message(chat_id: int, message_id: int) -> bool:
+    API_URL = f"https://api.telegram.org/bot{BOT_TOKEN}/deleteMessage"
+    payload = {
+        "chat_id": chat_id,
+        "message_id": message_id,
+    }
+    # Telegram expects a POST request with JSON body
+    with httpx.Client() as client:
+        response = client.post(API_URL, json=payload, timeout=10)
+    # The API returns JSON like {"ok":true,"result":true}
+    data = response.json()
+    return data.get("ok") and data.get("result") is True
+
+@celery_app_tg.task(
+    bind=True,
+    rate_limit='60/m',
+    name='tg.delete_message',
+    retry_backoff=True,
+    retry_backoff_max=600,
+    retry_jitter=True,
+    max_retries=20,
+    soft_time_limit=600,
+    time_limit=700,
+)
+def delete_from_tg(self, message_id: str, social: dict):
+    try:
+        chat_id = social.get('screen_name')
+        logging.info(f"TG удаление публикации {chat_id} {message_id}")
+        result = asyncio.run(delete_message(chat_id, message_id))
+        # result['doc'] = social
+        # httpx.post(f'{FULL_PATH_DOMAIN}/tg', json={'item': item, 'result': result}, headers={"Authorization": f"bearer {ATOKEN}"})
+    except Exception as exc:
+        raise self.retry(exc=exc, countdown=30)
+
+
 @celery_app_tg.task(
     bind=True,
     rate_limit='60/m',

vk.ipynb

@@ -331,14 +331,7 @@
     "    post_id = data[\"post_id\"]\n",
     "    doc = data[\"doc\"]\n",
     "\n",
-    "    db_name = item.get(\"db\")\n",
-    "    if db_name == \"competitions\":\n",
-    "        social = await social_collection.find_one({\"service\": True, \"type\": \"vk\"})\n",
-    "    else:\n",
-    "        social = doc\n",
-    "\n",
     "    logging.info(doc)\n",
-    "    logging.info(social)\n",
     "\n",
     "    owner_id = f\"-{doc.get('id')}\" if \"group\" in doc else doc.get(\"id\")\n",
     "\n",
@@ -359,8 +352,9 @@
     "        return {\"ok\": False, \"error\": res[\"error\"]}\n",
     "\n",
     "    # ---- Mongo remove vk post record ----\n",
+    "    if item:\n",
     "        col = item.get(\"col\")\n",
-    "    mongo_col = mongo_client[db_name][col]\n",
+    "        mongo_col = mongo_client[item.get(\"db\")][col]\n",
     "\n",
     "        mongo_col.update_one(\n",
     "            {\"_id\": ObjectId(item.get(\"_id\"))},\n",
@@ -423,9 +417,9 @@
     "    soft_time_limit=60,\n",
     "    time_limit=100,\n",
     ")\n",
-    "def delete_wall(self, item: dict, post_id: int, doc: dict):\n",
+    "def delete_wall(self, post_id: int, doc: dict, item: dict|None = None):\n",
     "    try:\n",
-    "        logging.info(f\"VK удаления публикации {item} {post_id}\")\n",
+    "        logging.info(f\"VK удаления публикации {doc} {post_id}\")\n",
     "        res = httpx.post(\n",
     "            'http://127.0.0.1:30080/api/v1/social/vk/delete',\n",
     "            headers={\"Authorization\": f\"bearer {ATOKEN}\"},\n",

vk.py

@@ -247,14 +247,7 @@ async def wall_delete(data: dict, request: Request):
     post_id = data["post_id"]
     doc = data["doc"]
 
-    db_name = item.get("db")
-    if db_name == "competitions":
-        social = await social_collection.find_one({"service": True, "type": "vk"})
-    else:
-        social = doc
-
     logging.info(doc)
-    logging.info(social)
 
     owner_id = f"-{doc.get('id')}" if "group" in doc else doc.get("id")
 
@@ -275,8 +268,9 @@ async def wall_delete(data: dict, request: Request):
         return {"ok": False, "error": res["error"]}
 
     # ---- Mongo remove vk post record ----
+    if item:
         col = item.get("col")
-    mongo_col = mongo_client[db_name][col]
+        mongo_col = mongo_client[item.get("db")][col]
 
         mongo_col.update_one(
             {"_id": ObjectId(item.get("_id"))},
@@ -325,9 +319,9 @@ def edit(self, post_id: int, item: dict, text: str, file_ids: list, social: dict
     soft_time_limit=60,
     time_limit=100,
 )
-def delete_wall(self, item: dict, post_id: int, doc: dict):
+def delete_wall(self, post_id: int, doc: dict, item: dict|None = None):
     try:
-        logging.info(f"VK удаления публикации {item} {post_id}")
+        logging.info(f"VK удаления публикации {doc} {post_id}")
         res = httpx.post(
             'http://127.0.0.1:30080/api/v1/social/vk/delete',
             headers={"Authorization": f"bearer {ATOKEN}"},