400 lines
17 KiB
Python
400 lines
17 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
SuperSam — Первая отправка согласования доставки (SMS)
|
||
Заменяет n8n workflow "Первая отправка согласования" (JnfZwOnpjEazseW5)
|
||
|
||
Логика:
|
||
1. SELECT order_groups WHERE status='ready_to_launch' AND delivery_status='pending_confirmation'
|
||
2. Для каждой группы с delivery_link и notification_status='link_ready':
|
||
a. Отправка SMS через sms.ru/sms/send
|
||
b. Ждёт wait_between_checks_seconds (25 сек по умолчанию)
|
||
c. Проверка статуса через sms.ru/sms/status
|
||
d. Если код 103 (доставлено) → UPDATE order_groups + лог
|
||
e. Если код 102 (в пути) → повторная проверка каждые wait_between_checks
|
||
f. Если в течение max_check_duration_minutes (90 мин) не доставлено → повторная отправка SMS
|
||
g. Если повторная отправка тоже не доставлена за max_check_duration → manual_required
|
||
3. Telegram-уведомление после каждой отправки
|
||
4. Полный лог в sms_campaign_log (видно в админке)
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import time
|
||
import json
|
||
import logging
|
||
import requests
|
||
from datetime import datetime, timezone, timedelta
|
||
from decimal import Decimal
|
||
|
||
import psycopg2
|
||
from psycopg2.extras import RealDictCursor
|
||
|
||
# ─── Конфигурация ────────────────────────────────────────────────────────────
|
||
|
||
DB_HOST = os.environ.get("DB_HOST", "10.0.4.5")
|
||
DB_PORT = os.environ.get("DB_PORT", "5432")
|
||
DB_NAME = os.environ.get("DB_NAME", "postgres")
|
||
DB_USER = os.environ.get("DB_USER", "supabase_admin")
|
||
DB_PASS = os.environ.get("DB_PASS", "4fe80bb21c7c3d17a8d8b226adf7a479")
|
||
|
||
TELEGRAM_BOT_TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN", "")
|
||
TELEGRAM_CHAT_ID = os.environ.get("TELEGRAM_CHAT_ID", "25164483")
|
||
|
||
SMS_API_ID = os.environ.get("SMS_API_ID", "C92063B3-95ED-8559-157B-1946EB5A2486")
|
||
SMS_SEND_URL = "https://sms.ru/sms/send"
|
||
SMS_STATUS_URL = "https://sms.ru/sms/status"
|
||
|
||
LOG_FILE = "/var/log/supersam-sms-first.log"
|
||
|
||
# ─── Логирование ─────────────────────────────────────────────────────────────
|
||
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format="%(asctime)s [%(levelname)s] %(message)s",
|
||
handlers=[
|
||
logging.FileHandler(LOG_FILE),
|
||
logging.StreamHandler(sys.stdout),
|
||
],
|
||
)
|
||
log = logging.getLogger("sms_first")
|
||
|
||
# ─── БД ──────────────────────────────────────────────────────────────────────
|
||
|
||
def get_db_conn():
|
||
return psycopg2.connect(
|
||
host=DB_HOST, port=DB_PORT, dbname=DB_NAME,
|
||
user=DB_USER, password=DB_PASS,
|
||
)
|
||
|
||
def load_settings(conn):
|
||
"""Загружает настройки из sms_campaign_settings (управляются мегаадмином)"""
|
||
with conn.cursor(cursor_factory=RealDictCursor) as cur:
|
||
cur.execute("SELECT * FROM sms_campaign_settings WHERE campaign_type = 'first_sms'")
|
||
row = cur.fetchone()
|
||
if not row:
|
||
return {
|
||
"wait_between_checks_seconds": 25,
|
||
"max_check_duration_minutes": 90,
|
||
"max_attempts": 2,
|
||
"enabled": True,
|
||
"telegram_chat_id": "25164483",
|
||
}
|
||
return dict(row)
|
||
|
||
def get_pending_groups(conn):
|
||
"""Получает группы для отправки первой SMS"""
|
||
with conn.cursor(cursor_factory=RealDictCursor) as cur:
|
||
cur.execute("""
|
||
SELECT id, group_key, customer_name, customer_phone, customer_phone_normalized,
|
||
delivery_link, notification_status, status, delivery_status
|
||
FROM order_groups
|
||
WHERE status = 'ready_to_launch'
|
||
AND delivery_status = 'pending_confirmation'
|
||
ORDER BY created_at ASC
|
||
""")
|
||
return [dict(row) for row in cur.fetchall()]
|
||
|
||
def update_order_group(conn, group_id, fields):
|
||
"""Обновляет order_groups"""
|
||
with conn.cursor() as cur:
|
||
set_parts = []
|
||
values = []
|
||
for k, v in fields.items():
|
||
if v == "NOW()":
|
||
set_parts.append(f"{k} = NOW()")
|
||
else:
|
||
set_parts.append(f"{k} = %s")
|
||
values.append(v)
|
||
values.append(group_id)
|
||
cur.execute(f"UPDATE order_groups SET {', '.join(set_parts)} WHERE id = %s", values)
|
||
conn.commit()
|
||
|
||
def insert_sms_log(conn, **kwargs):
|
||
"""Записывает лог в sms_campaign_log"""
|
||
with conn.cursor() as cur:
|
||
cols = ", ".join(kwargs.keys())
|
||
placeholders = ", ".join(["%s"] * len(kwargs))
|
||
cur.execute(f"INSERT INTO sms_campaign_log ({cols}) VALUES ({placeholders})", list(kwargs.values()))
|
||
conn.commit()
|
||
|
||
def update_sms_log(conn, log_id, **kwargs):
|
||
"""Обновляет запись лога"""
|
||
with conn.cursor() as cur:
|
||
set_parts = []
|
||
values = []
|
||
for k, v in kwargs.items():
|
||
set_parts.append(f"{k} = %s")
|
||
values.append(v)
|
||
values.append(log_id)
|
||
cur.execute(f"UPDATE sms_campaign_log SET {', '.join(set_parts)}, updated_at = NOW() WHERE id = %s", values)
|
||
conn.commit()
|
||
|
||
# ─── SMS API ─────────────────────────────────────────────────────────────────
|
||
|
||
def send_sms(phone, message, api_id):
|
||
"""Отправляет SMS через sms.ru, возвращает (sms_id, raw_response)"""
|
||
try:
|
||
# Нормализуем телефон: убираем всё кроме цифр, добавляем 7
|
||
clean_phone = "".join(c for c in phone if c.isdigit())
|
||
if clean_phone.startswith("8"):
|
||
clean_phone = "7" + clean_phone[1:]
|
||
elif not clean_phone.startswith("7"):
|
||
clean_phone = "7" + clean_phone
|
||
|
||
resp = requests.post(SMS_SEND_URL, params={
|
||
"api_id": api_id,
|
||
"to": clean_phone,
|
||
}, data={"msg": message}, timeout=30)
|
||
text = resp.text
|
||
log.info(f"SMS send response: {text[:200]}")
|
||
|
||
# Парсим sms_id: sms.ru возвращает строки, вторая строка = sms_id
|
||
lines = text.strip().split("\n")
|
||
if len(lines) >= 2 and lines[0].strip() == "100":
|
||
sms_id = lines[1].strip()
|
||
return sms_id, text
|
||
elif lines and lines[0].strip() == "100":
|
||
sms_id = lines[1].strip() if len(lines) > 1 else ""
|
||
return sms_id, text
|
||
else:
|
||
return None, text
|
||
except Exception as e:
|
||
log.error(f"SMS send error: {e}")
|
||
return None, str(e)
|
||
|
||
def check_sms_status(sms_id, api_id):
|
||
"""Проверяет статус SMS, возвращает код (102=в пути, 103=доставлено)"""
|
||
try:
|
||
resp = requests.post(SMS_STATUS_URL, params={
|
||
"api_id": api_id,
|
||
"sms_id": sms_id,
|
||
}, timeout=30)
|
||
text = resp.text
|
||
log.info(f"SMS status response for {sms_id}: {text[:200]}")
|
||
|
||
lines = text.strip().split("\n")
|
||
if len(lines) >= 2:
|
||
code = lines[1].strip()
|
||
return code, text
|
||
return None, text
|
||
except Exception as e:
|
||
log.error(f"SMS status check error: {e}")
|
||
return None, str(e)
|
||
|
||
# ─── Telegram ────────────────────────────────────────────────────────────────
|
||
|
||
def send_telegram(message, chat_id):
|
||
"""Отправляет сообщение в Telegram"""
|
||
if not TELEGRAM_BOT_TOKEN:
|
||
log.warning("TELEGRAM_BOT_TOKEN not set, skipping Telegram notification")
|
||
return
|
||
try:
|
||
resp = requests.post(
|
||
f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage",
|
||
json={"chat_id": chat_id, "text": message, "parse_mode": "HTML"},
|
||
timeout=10,
|
||
)
|
||
if resp.status_code != 200:
|
||
log.warning(f"Telegram error: {resp.text[:200]}")
|
||
except Exception as e:
|
||
log.warning(f"Telegram send failed: {e}")
|
||
|
||
# ─── Основная логика ─────────────────────────────────────────────────────────
|
||
|
||
def process_group(conn, group, settings):
|
||
"""Обрабатывает одну группу: отправка SMS → проверка доставки → UPDATE"""
|
||
group_id = group["id"]
|
||
customer_name = group.get("customer_name") or group.get("group_key", "—")
|
||
phone = group.get("customer_phone_normalized") or group.get("customer_phone", "")
|
||
delivery_link = group.get("delivery_link", "")
|
||
|
||
if not delivery_link:
|
||
log.warning(f"Group {group_id}: no delivery_link, skipping")
|
||
return False
|
||
|
||
if group.get("notification_status") != "link_ready":
|
||
log.info(f"Group {group_id}: notification_status={group.get('notification_status')}, skipping")
|
||
return False
|
||
|
||
api_id = settings.get("sms_api_id", SMS_API_ID)
|
||
wait_sec = settings.get("wait_between_checks_seconds", 25)
|
||
max_duration_min = settings.get("max_check_duration_minutes", 90)
|
||
max_attempts = settings.get("max_attempts", 2)
|
||
tg_chat = settings.get("telegram_chat_id", TELEGRAM_CHAT_ID)
|
||
|
||
sms_text = f"Ваш заказ готов. Согласуйте дату доставки по ссылке: {delivery_link}"
|
||
|
||
log.info(f"=== Group {group_id} ({customer_name}, phone={phone}) ===")
|
||
|
||
for attempt in range(1, max_attempts + 1):
|
||
log.info(f"Group {group_id}: attempt {attempt}/{max_attempts}")
|
||
|
||
# Запись лога отправки
|
||
log_id = None
|
||
with conn.cursor(cursor_factory=RealDictCursor) as cur:
|
||
cur.execute("""
|
||
INSERT INTO sms_campaign_log
|
||
(campaign_type, order_group_id, customer_phone, sms_text, status, attempts, created_at)
|
||
VALUES ('first_sms', %s, %s, %s, 'sending', %s, NOW())
|
||
RETURNING id
|
||
""", (str(group_id), phone, sms_text, attempt))
|
||
log_id = cur.fetchone()["id"]
|
||
conn.commit()
|
||
|
||
# Отправка SMS
|
||
sms_id, raw = send_sms(phone, sms_text, api_id)
|
||
|
||
if not sms_id:
|
||
error = raw[:500] if raw else "No sms_id in response"
|
||
log.error(f"Group {group_id}: SMS send failed: {error}")
|
||
update_sms_log(conn, log_id, status="send_failed", sms_code=None, error_message=error)
|
||
if attempt < max_attempts:
|
||
log.info(f"Group {group_id}: retrying in {wait_sec}s...")
|
||
time.sleep(wait_sec)
|
||
continue
|
||
else:
|
||
update_order_group(conn, group_id, {
|
||
"notification_status": "send_failed",
|
||
"last_sms_error": error[:200],
|
||
"status": "ready_to_launch", # остаётся для повторной попытки
|
||
})
|
||
send_telegram(f"❌ SMS не отправлена: {customer_name} ({phone})\nОшибка: {error[:200]}", tg_chat)
|
||
return False
|
||
|
||
# SMS отправлена успешно
|
||
update_sms_log(conn, log_id, status="sent", sms_id=sms_id, sms_code="100")
|
||
log.info(f"Group {group_id}: SMS sent, sms_id={sms_id}")
|
||
|
||
# Цикл проверки доставки (до max_check_duration_minutes)
|
||
start_time = datetime.now(timezone.utc)
|
||
max_duration = timedelta(minutes=max_duration_min)
|
||
check_interval = wait_sec
|
||
|
||
while datetime.now(timezone.utc) - start_time < max_duration:
|
||
log.info(f"Group {group_id}: waiting {check_interval}s before status check...")
|
||
time.sleep(check_interval)
|
||
|
||
code, raw_status = check_sms_status(sms_id, api_id)
|
||
update_sms_log(conn, log_id, status="checking", sms_code=code)
|
||
|
||
if code == "103":
|
||
# Доставлено!
|
||
log.info(f"Group {group_id}: SMS delivered (code=103)!")
|
||
update_sms_log(conn, log_id, status="delivered", sms_code="103")
|
||
|
||
now_iso = datetime.now(timezone.utc).isoformat()
|
||
update_order_group(conn, group_id, {
|
||
"notification_status": "first_sms_sent",
|
||
"sms_attempts": attempt,
|
||
"first_sms_sent_at": "NOW()",
|
||
"last_sms_error": None,
|
||
"sms_sent_at": "NOW()",
|
||
"next_notification_check_at": (datetime.now(timezone.utc) + timedelta(hours=settings.get("second_sms_delay_hours", 3))).isoformat(),
|
||
"status": "first_sms_sent",
|
||
})
|
||
|
||
send_telegram(
|
||
f"✅ SMS доставлена: {customer_name} ({phone})\nПопытка {attempt}\nsms_id: {sms_id}",
|
||
tg_chat,
|
||
)
|
||
return True
|
||
|
||
elif code == "102":
|
||
# В пути — продолжаем ждать
|
||
log.info(f"Group {group_id}: SMS in transit (code=102), continuing to check...")
|
||
continue
|
||
|
||
elif code in ("104", "105", "106", "107", "108"):
|
||
# Ошибка доставки
|
||
error_msg = f"SMS error code: {code}"
|
||
log.error(f"Group {group_id}: {error_msg}")
|
||
update_sms_log(conn, log_id, status="error", sms_code=code, error_message=error_msg)
|
||
# Прерываем внутренний цикл, переходим к следующей попытке
|
||
break
|
||
|
||
else:
|
||
# Неизвестный код — продолжаем проверять
|
||
log.warning(f"Group {group_id}: unknown SMS code: {code}")
|
||
continue
|
||
|
||
# Время проверки истекло или ошибка — повторная отправка
|
||
elapsed = (datetime.now(timezone.utc) - start_time).total_seconds() / 60
|
||
log.warning(f"Group {group_id}: check duration exceeded ({elapsed:.0f} min) or error, attempt {attempt}")
|
||
update_sms_log(conn, log_id, status="timeout", error_message=f"Not delivered in {max_duration_min} min")
|
||
|
||
if attempt < max_attempts:
|
||
log.info(f"Group {group_id}: will retry SMS send...")
|
||
continue
|
||
else:
|
||
# Все попытки исчерпаны
|
||
update_order_group(conn, group_id, {
|
||
"notification_status": "send_failed",
|
||
"last_sms_error": f"Not delivered after {max_attempts} attempts",
|
||
"status": "ready_to_launch",
|
||
})
|
||
send_telegram(
|
||
f"⚠️ SMS не доставлена после {max_attempts} попыток: {customer_name} ({phone})\nТребуется ручное управление",
|
||
tg_chat,
|
||
)
|
||
return False
|
||
|
||
return False
|
||
|
||
|
||
def main():
|
||
log.info("=" * 60)
|
||
log.info("SMS First Campaign — START")
|
||
conn = get_db_conn()
|
||
|
||
try:
|
||
settings = load_settings(conn)
|
||
log.info(f"Settings: wait={settings.get('wait_between_checks_seconds')}s, "
|
||
f"max_duration={settings.get('max_check_duration_minutes')}min, "
|
||
f"max_attempts={settings.get('max_attempts')}")
|
||
|
||
if not settings.get("enabled", True):
|
||
log.info("Campaign disabled, exiting")
|
||
return
|
||
|
||
groups = get_pending_groups(conn)
|
||
log.info(f"Found {len(groups)} groups with status=ready_to_launch, delivery_status=pending_confirmation")
|
||
|
||
if not groups:
|
||
log.info("No groups to process, exiting")
|
||
return
|
||
|
||
tg_chat = settings.get("telegram_chat_id", TELEGRAM_CHAT_ID)
|
||
sent = 0
|
||
delivered = 0
|
||
failed = 0
|
||
|
||
for group in groups:
|
||
try:
|
||
result = process_group(conn, group, settings)
|
||
if result:
|
||
delivered += 1
|
||
else:
|
||
failed += 1
|
||
sent += 1
|
||
except Exception as e:
|
||
log.error(f"Error processing group {group.get('id')}: {e}", exc_info=True)
|
||
failed += 1
|
||
|
||
# Сводка в Telegram
|
||
summary = (f"📊 <b>Первая отправка — сводка</b>\n\n"
|
||
f"Групп найдено: {len(groups)}\n"
|
||
f"Отправлено: {sent}\n"
|
||
f"Доставлено: {delivered}\n"
|
||
f"Ошибок: {failed}")
|
||
send_telegram(summary, tg_chat)
|
||
log.info(f"Summary: sent={sent}, delivered={delivered}, failed={failed}")
|
||
|
||
finally:
|
||
conn.close()
|
||
|
||
log.info("SMS First Campaign — END")
|
||
log.info("=" * 60)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main() |