59 lines
2.5 KiB
Python
59 lines
2.5 KiB
Python
"""Hooks SoloBot's senders into the Cosmo App inbox. Idempotent; backs up files first; restores them on failure.
|
|
Run with the bot's venv python on the bot server: python apply_patch.py /root/cosmonetvpn_bot /root/backup"""
|
|
import py_compile
|
|
import shutil
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
bot, backup = Path(sys.argv[1]), Path(sys.argv[2])
|
|
IMPORT = "from cosmo_app_inbox import deliver_to_app, is_app_user\n"
|
|
PATCHES = {
|
|
"handlers/admin/sender/sender_service.py": [
|
|
("from logger import logger\n", "from logger import logger\n" + IMPORT),
|
|
(" async def _send_single_message(self, msg: BroadcastMessage) -> bool:\n try:\n",
|
|
" async def _send_single_message(self, msg: BroadcastMessage) -> bool:\n"
|
|
" if is_app_user(msg.tg_id): # cosmo_app_inbox: app user without Telegram -> app inbox\n"
|
|
" return await deliver_to_app(msg.tg_id, msg.text)\n"
|
|
" try:\n"),
|
|
],
|
|
"handlers/notifications/notify_utils.py": [
|
|
("from logger import logger\n", "from logger import logger\n" + IMPORT),
|
|
(" async def _send_single_message(self, msg: NotificationMessage) -> bool:\n try:\n",
|
|
" async def _send_single_message(self, msg: NotificationMessage) -> bool:\n"
|
|
" if is_app_user(msg.tg_id): # cosmo_app_inbox: the app shows its own reminders\n"
|
|
" return True\n"
|
|
" try:\n"),
|
|
(") -> bool:\n if image_filename is None:\n",
|
|
") -> bool:\n"
|
|
" if is_app_user(tg_id): # cosmo_app_inbox: the app shows its own reminders\n"
|
|
" return True\n"
|
|
" if image_filename is None:\n"),
|
|
],
|
|
}
|
|
|
|
ts = time.strftime("%Y-%m-%d-%H%M")
|
|
done = []
|
|
try:
|
|
for rel, repl in PATCHES.items():
|
|
p = bot / rel
|
|
src = p.read_text(encoding="utf-8")
|
|
if "cosmo_app_inbox" in src:
|
|
print(f"{rel}: already patched")
|
|
continue
|
|
bak = backup / f"{p.name}.before-cosmo-app.{ts}"
|
|
shutil.copy2(p, bak)
|
|
done.append((p, bak))
|
|
for old, new in repl:
|
|
n = src.count(old)
|
|
if n != 1:
|
|
raise RuntimeError(f"{rel}: anchor found {n} times: {old[:60]!r}")
|
|
src = src.replace(old, new)
|
|
p.write_text(src, encoding="utf-8")
|
|
py_compile.compile(str(p), doraise=True)
|
|
print(f"{rel}: patched (backup {bak})")
|
|
except Exception as e:
|
|
for p, bak in done:
|
|
shutil.copy2(bak, p)
|
|
print(f"FAILED, restored originals: {e}")
|
|
sys.exit(1)
|