supersam/scripts/sms_paid_storage_campaign.py

608 lines
27 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env python3
"""
SuperSam — Paid Storage Campaign (State Machine)
Уведомление о включении платного хранения.
Логика:
Когда delivery_status = 'paid_storage' (логист/админ включил):
→ Если рабочие часы (8-21, Пн-Пт): отправить SMS один раз
→ Если ночью: ждать до утра
→ Проверить доставку (103)
→ Пометить notification_status = 'paid_storage_sent'
SMS текст: "Ваш заказ переведён на платное хранение. Стоимость: 300 ₽/день.
Заберите заказ или согласуйте доставку: {link}"
Защита от повторной отправки:
1. После отправки → notification_status = 'paid_storage_sending'
2. EXISTS проверка в sms_campaign_log за 24h
"""
import os
import sys
import logging
from datetime import datetime, timezone, timedelta
import requests
import psycopg2
from psycopg2.extras import RealDictCursor
try: # лимитер sms.ru (модуль лежит рядом со скриптом)
from sms_rate_limit import (
check_send_allowed, alert_due, acquire_send_lock, release_send_lock,
)
except ImportError: # noqa: BLE001 — без лимитера кампания работает как раньше
check_send_allowed = None
alert_due = None
acquire_send_lock = None
release_send_lock = None
# ─── Конфигурация ────────────────────────────────────────────────────────────
DB_HOST = os.environ.get("DB_HOST", "10.0.4.12")
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-paid-storage.log"
CAMPAIGN_TYPE = "paid_storage"
IN_TRANSIT_CODES = {"100", "101", "102"}
DELIVERED_CODE = "103"
DELIVERY_ERROR_CODES = {"104", "105", "106", "107", "108", "130"}
LIMIT_ERROR_CODES = {"131", "132", "230", "231", "232"}
# ─── Тексты SMS ──────────────────────────────────────────────────────────────
# Шаблоны живут в админке (sms_campaign_settings):
# sms_text_template — текст для ДОСТАВКИ
# sms_text_template_pickup — текст для САМОВЫВОЗА
# Подстановки: {link}/{ссылка} — ссылка согласования,
# {order}/{номер}/{номер_доставки} — номер доставки (010796),
# {code}/{код}/{код_выдачи} — код выдачи (7598).
# Пустой шаблон → DEFAULT_TEXT_*.
DEFAULT_TEXT_DELIVERY = (
"Ваш заказ переведён на платное хранение. Стоимость: 300 ₽/день. "
"Заберите заказ или согласуйте доставку: {link}"
)
DEFAULT_TEXT_PICKUP = (
"Заказ переведён на платное хранение. Номер доставки {order}, код выдачи {code}. "
"Заберите со склада или согласуйте дату самовывоза: {link}"
)
def first_order_digits(value):
"""Номер доставки из order_numbers («ТП\\-010722» → «010722»)."""
items = value if isinstance(value, (list, tuple)) else [value]
for item in items:
digits = "".join(ch for ch in str(item or "").replace("\\", "") if ch.isdigit())
if digits:
return digits
return ""
def apply_sms_placeholders(text, group, link):
"""Подстановка переменных шаблона в текст SMS."""
if not text:
return text
order = first_order_digits(group.get("order_numbers"))
code = "".join(ch for ch in str(group.get("pickup_code") or "") if ch.isdigit())
replacements = (
("{link}", link), ("{ссылка}", link),
("{номер_доставки}", order), ("{номер}", order), ("{order}", order),
("{код_выдачи}", code), ("{код}", code), ("{code}", code),
)
result = text
for token, value in replacements:
result = result.replace(token, value)
return result.strip()
def build_sms_text(group, settings):
"""Текст SMS о платном хранении по типу отгрузки."""
link = group.get("delivery_link") or ""
pickup = (group.get("delivery_type") or "delivery").strip().lower() == "pickup"
template = (settings.get(
"sms_text_template_pickup" if pickup else "sms_text_template"
) or "").strip()
if not template:
template = DEFAULT_TEXT_PICKUP if pickup else DEFAULT_TEXT_DELIVERY
return apply_sms_placeholders(template, group, link)
# ─── Логирование ─────────────────────────────────────────────────────────────
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_paid_storage")
# ─── БД ──────────────────────────────────────────────────────────────────────
def get_db_conn():
return psycopg2.connect(
host=DB_HOST, port=DB_PORT, dbname=DB_NAME,
user=DB_USER, password=DB_PASS,
)
def apply_schedule_work_days(conn, settings):
"""Дни отправки — из «Графика работы» в ЛК (единый источник правды).
Если график прочитать не удалось, остаются прежние дни кампании.
"""
try:
with conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute("SELECT sms_days FROM business_schedule_settings WHERE id = 1")
row = cur.fetchone() or {}
days = row.get("sms_days") or []
if days:
settings["work_days"] = ",".join(str(int(d)) for d in days)
except Exception as exc: # noqa: BLE001 — не роняем рассылку из-за графика
print(f"График работы не прочитан, дни SMS из настроек кампании: {exc}")
return settings
def load_settings(conn):
with conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute("SELECT * FROM sms_campaign_settings WHERE campaign_type = %s", (CAMPAIGN_TYPE,))
row = cur.fetchone()
if not row:
return {
"enabled": True, "test_mode": True, "test_phone": "79788382260",
"max_check_duration_minutes": 90, "max_attempts": 2,
"telegram_chat_id": TELEGRAM_CHAT_ID, "sms_api_id": SMS_API_ID,
"work_hours_start": 8, "work_hours_end": 21, "work_days": "1,2,3,4,5",
}
settings = dict(row)
return apply_schedule_work_days(conn, settings)
# ─── SMS API ─────────────────────────────────────────────────────────────────
def normalize_phone(phone):
clean = "".join(c for c in str(phone) if c.isdigit())
if clean.startswith("8"):
clean = "7" + clean[1:]
elif not clean.startswith("7"):
clean = "7" + clean
return clean
def send_sms(phone, message, api_id):
try:
clean_phone = normalize_phone(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 to {clean_phone}: {text[:200]}")
lines = text.strip().split("\n")
status_code = lines[0].strip() if lines else ""
if status_code == "100" and len(lines) >= 2:
return lines[1].strip(), text, "100"
return None, text, status_code
except Exception as e:
log.error(f"SMS send error: {e}")
return None, str(e), "error"
def check_sms_status(sms_id, api_id):
try:
resp = requests.post(SMS_STATUS_URL, params={"api_id": api_id, "sms_id": sms_id}, timeout=30)
text = resp.text
lines = text.strip().split("\n")
status_code = lines[0].strip() if lines else ""
sms_status_code = lines[1].strip() if len(lines) >= 2 else None
log.info(f"SMS status for {sms_id}: code={status_code}, sms_status={sms_status_code}")
return sms_status_code, text, status_code
except Exception as e:
log.error(f"SMS status check error: {e}")
return None, str(e), "error"
def fetch_balance(api_id):
try:
resp = requests.get("https://sms.ru/my/balance", params={"api_id": api_id}, timeout=15)
text = resp.text
lines = text.strip().split("\n")
if lines[0].strip() == "100" and len(lines) >= 2:
return float(lines[1].strip()), text
return None, text
except Exception as e:
log.error(f"Balance fetch error: {e}")
return None, str(e)
# ─── Telegram ────────────────────────────────────────────────────────────────
def send_telegram(message, chat_id):
if not TELEGRAM_BOT_TOKEN:
log.warning("TELEGRAM_BOT_TOKEN not set, skipping Telegram")
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 failed: {e}")
# ─── Проверка рабочего времени ────────────────────────────────────────────────
def is_within_work_hours(settings):
now_msk = datetime.now(timezone(timedelta(hours=3)))
today_num = now_msk.weekday() + 1
allowed_days = set()
work_days_str = settings.get("work_days", "1,2,3,4,5")
for part in str(work_days_str).split(","):
part = part.strip()
if part.isdigit():
allowed_days.add(int(part))
if today_num not in allowed_days:
return False
hour = now_msk.hour
start_h = settings.get("work_hours_start", 8)
end_h = settings.get("work_hours_end", 21)
return start_h <= hour < end_h
# ─── State Machine ───────────────────────────────────────────────────────────
def get_groups_to_send(conn):
"""Группы с paid_storage, которым ещё не отправили уведомление."""
with conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute("""
SELECT og.id, og.group_key, og.customer_name, og.customer_phone,
og.customer_phone_normalized, og.delivery_link, og.notification_status,
og.paid_storage_at, og.delivery_type, og.town,
og.order_numbers, og.pickup_code
FROM order_groups og
LEFT JOIN sms_city_profiles scp ON scp.city = og.town
WHERE og.delivery_status = 'paid_storage'
AND og.paid_storage_at IS NOT NULL
-- Гео-исключения — общие для SMS и звонков
AND NOT EXISTS (
SELECT 1 FROM sms_city_rules r
WHERE r.city = og.town
AND r.enabled = false
AND r.campaign_type IN ('*', 'paid_storage')
AND (r.valid_from IS NULL OR r.valid_from <= CURRENT_DATE)
AND (r.valid_until IS NULL OR r.valid_until >= CURRENT_DATE)
)
AND COALESCE(og.notification_status, '') NOT IN ('paid_storage_sending', 'paid_storage_sent')
AND og.delivery_link IS NOT NULL
AND og.delivery_link != ''
-- Профиль: paid_storage_enabled (если false — SMS о платном хранении не отправляется)
AND COALESCE(scp.paid_storage_enabled, true) = true
AND NOT EXISTS (
SELECT 1 FROM sms_campaign_log scl
WHERE scl.order_group_id = og.id
AND scl.campaign_type = 'paid_storage'
AND scl.status IN ('sent', 'checking')
AND scl.created_at > NOW() - INTERVAL '24 hours'
)
ORDER BY og.paid_storage_at ASC
""")
return [dict(r) for r in cur.fetchall()]
def get_sms_to_check(conn, max_duration_min):
with conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute("""
SELECT scl.id as log_id, scl.sms_id, scl.order_group_id, scl.customer_phone,
scl.attempts, scl.created_at, scl.sms_code, scl.needs_check,
og.customer_name, og.group_key
FROM sms_campaign_log scl
JOIN order_groups og ON og.id = scl.order_group_id
WHERE scl.campaign_type = 'paid_storage'
AND scl.status IN ('sent', 'checking')
AND scl.sms_id IS NOT NULL
AND (scl.created_at > NOW() - INTERVAL '%s minutes' OR scl.needs_check = true)
ORDER BY scl.needs_check DESC, scl.created_at ASC
""" % max_duration_min)
return [dict(r) for r in cur.fetchall()]
def get_sms_expired(conn, max_duration_min):
with conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute("""
SELECT scl.id as log_id, scl.sms_id, scl.order_group_id, scl.customer_phone,
scl.attempts, scl.created_at,
og.customer_name, og.group_key, og.delivery_link
FROM sms_campaign_log scl
JOIN order_groups og ON og.id = scl.order_group_id
WHERE scl.campaign_type = 'paid_storage'
AND scl.status IN ('sent', 'checking')
AND scl.sms_id IS NOT NULL
AND scl.created_at < NOW() - INTERVAL '%s minutes'
ORDER BY scl.created_at ASC
""" % max_duration_min)
return [dict(r) for r in cur.fetchall()]
def insert_sms_log(conn, **kwargs):
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}) RETURNING id", list(kwargs.values()))
log_id = cur.fetchone()[0]
conn.commit()
return log_id
def update_sms_log(conn, log_id, **kwargs):
with conn.cursor() as cur:
set_parts = []
values = []
for k, v in kwargs.items():
if v == "NOW()":
set_parts.append(f"{k} = NOW()")
else:
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()
def update_order_group(conn, group_id, fields):
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 step_send_new(conn, settings):
api_id = settings.get("sms_api_id", SMS_API_ID)
tg_chat = settings.get("telegram_chat_id", TELEGRAM_CHAT_ID)
groups = get_groups_to_send(conn)
log.info(f"Step 1: {len(groups)} groups to send paid_storage SMS")
# --- Сериализация отправки: аккаунтный лимит sms.ru общий для всех кампаний ---
_send_lock = False
if acquire_send_lock is not None:
_send_lock = acquire_send_lock(conn)
if not _send_lock:
log.warning("Прогон пропущен: не удалось взять блокировку отправки")
return 0
sent_count = 0
limited_count = 0
limited_last = None
limited_info = None
for group in groups:
group_id = str(group["id"])
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", "")
with conn.cursor() as cur:
cur.execute("""
SELECT COUNT(*) FROM sms_campaign_log
WHERE order_group_id = %s AND campaign_type = 'paid_storage'
AND created_at > NOW() - INTERVAL '24 hours'
AND status IN ('sent', 'checking', 'delivered')
""", (group_id,))
if cur.fetchone()[0] > 0:
log.info(f"Group {group_id}: already has recent paid_storage SMS, skipping")
continue
sms_text = build_sms_text(group, settings)
log.info(f"Group {group_id}: paid_storage SMS ({group.get('delivery_type') or 'delivery'}): {sms_text!r}")
send_phone = phone
if settings.get("test_mode", True):
send_phone = settings.get("test_phone", "79788382260")
log.info(f"TEST MODE: sending to {send_phone} instead of {phone}")
# --- Лимит sms.ru: не больше N SMS на один номер за окно (иначе блокировка) ---
if check_send_allowed is not None:
_ok, _reason, _info = check_send_allowed(conn, send_phone, settings)
if not _ok:
limited_count += 1
limited_last = _reason
limited_info = _info
log.warning(f"Group {group_id}: SMS HELD by sms.ru rate limit - {_reason}")
continue # сообщение не потеряно: уйдёт в следующий запуск
log.info(f"Sending paid_storage SMS to {name} (orig={phone}, send={send_phone})")
sms_id, raw, code = send_sms(send_phone, sms_text, api_id)
if sms_id:
log_id = insert_sms_log(conn,
campaign_type=CAMPAIGN_TYPE,
order_group_id=group_id,
customer_phone=phone,
sent_to=send_phone,
was_test_mode=bool(settings.get("test_mode", True)),
sms_id=sms_id,
sms_text=sms_text,
status="sent",
sms_code=code,
attempts=1,
)
update_order_group(conn, group_id, {
"notification_status": "paid_storage_sending",
"sms_sent_at": "NOW()",
})
log.info(f"Group {group_id}: paid_storage SMS sent, sms_id={sms_id}, notification_status→paid_storage_sending")
sent_count += 1
else:
error = raw[:500] if raw else "Unknown error"
insert_sms_log(conn,
campaign_type=CAMPAIGN_TYPE,
order_group_id=group_id,
customer_phone=phone,
sms_text=sms_text,
status="send_failed",
sms_code=code,
attempts=1,
error_message=error,
)
log.error(f"Group {group_id}: paid_storage SMS failed (code={code}): {error[:200]}")
update_order_group(conn, group_id, {"last_sms_error": error[:200]})
send_telegram(f"❌ SMS платное хранение не отправлена: {name} ({phone})\nКод: {code}", tg_chat)
if limited_count:
_win = (limited_info or {}).get("window")
_cap = (limited_info or {}).get("per_number")
_acct = (limited_info or {}).get("account")
msg = (
f"SMS придержаны лимитом sms.ru: {limited_count} шт. "
f"(окно {_win}с, лимит аккаунта {_acct} суммарно, "
f"страховка {_cap} на номер); "
f"причина: {limited_last}; "
f"сообщения не потеряны — уйдут в следующий запуск."
)
log.warning("Rate limit held %s SMS: %s", limited_count, limited_last)
if alert_due is not None and alert_due(conn, settings):
send_telegram(msg, tg_chat)
else:
log.info("Rate limit Telegram alert suppressed (throttled)")
if _send_lock and release_send_lock is not None:
release_send_lock(conn)
return sent_count
def step_check_status(conn, settings):
api_id = settings.get("sms_api_id", SMS_API_ID)
tg_chat = settings.get("telegram_chat_id", TELEGRAM_CHAT_ID)
max_duration = settings.get("max_check_duration_minutes", 90)
sms_list = get_sms_to_check(conn, max_duration)
log.info(f"Step 2: {len(sms_list)} paid_storage SMS to check")
delivered = 0
for item in sms_list:
log_id = item["log_id"]
sms_id = item["sms_id"]
group_id = str(item["order_group_id"])
name = item.get("customer_name") or item.get("group_key", "—")
phone = item.get("customer_phone", "")
code, raw, api_code = check_sms_status(sms_id, api_id)
update_sms_log(conn, log_id, needs_check=False, checked_at="NOW()")
if code == DELIVERED_CODE:
log.info(f"Group {group_id}: paid_storage SMS delivered (103)!")
update_sms_log(conn, log_id, status="delivered", sms_code=code)
update_order_group(conn, group_id, {"notification_status": "paid_storage_sent"})
send_telegram(f"✅ SMS платное хранение доставлена: {name} ({phone})", tg_chat)
delivered += 1
elif code in IN_TRANSIT_CODES:
log.info(f"Group {group_id}: in transit (code={code})")
update_sms_log(conn, log_id, status="checking", sms_code=code)
elif code in DELIVERY_ERROR_CODES:
log.error(f"Group {group_id}: delivery error (code={code})")
update_sms_log(conn, log_id, status="error", sms_code=code, error_message=f"Delivery error: {code}")
update_sms_log(conn, log_id, status="expired")
# Сброс для retry
update_order_group(conn, group_id, {"notification_status": "manual_required"})
elif code in LIMIT_ERROR_CODES:
log.error(f"Group {group_id}: limit exceeded (code={code})")
update_sms_log(conn, log_id, status="limit_exceeded", sms_code=code, error_message=f"Limit: {code}")
update_order_group(conn, group_id, {
"notification_status": "manual_required",
"last_sms_error": f"Limit exceeded (code={code})",
})
send_telegram(f"🚫 SMS платное хранение заблокирована (лимит): {name} ({phone})", tg_chat)
else:
log.warning(f"Group {group_id}: unknown code: {code}")
update_sms_log(conn, log_id, status="checking", sms_code=code, error_message=f"Unknown: {code}")
return delivered
def step_handle_expired(conn, settings):
max_duration = settings.get("max_check_duration_minutes", 90)
tg_chat = settings.get("telegram_chat_id", TELEGRAM_CHAT_ID)
expired = get_sms_expired(conn, max_duration)
log.info(f"Step 3: {len(expired)} paid_storage SMS expired")
for item in expired:
log_id = item["log_id"]
group_id = str(item["order_group_id"])
name = item.get("customer_name") or item.get("group_key", "—")
phone = item.get("customer_phone", "")
log.warning(f"Group {group_id}: paid_storage SMS expired")
update_sms_log(conn, log_id, status="expired", error_message=f"Not delivered in {max_duration} min")
update_order_group(conn, group_id, {"notification_status": "paid_storage_sent"})
send_telegram(f"⚠️ SMS платное хранение не доставлена: {name} ({phone})", tg_chat)
# ─── Main ────────────────────────────────────────────────────────────────────
def main():
log.info("=" * 60)
log.info("Paid Storage Campaign — START")
conn = get_db_conn()
try:
settings = load_settings(conn)
log.info(f"Settings: work={settings.get('work_hours_start')}-{settings.get('work_hours_end')}, "
f"days={settings.get('work_days')}, test={settings.get('test_mode')}")
if not settings.get("enabled", True):
log.info("Campaign disabled, exiting")
return
# Отправка только в рабочие часы
work_hours = is_within_work_hours(settings)
sent = 0
if work_hours:
sent = step_send_new(conn, settings)
else:
log.info("Outside work hours, skipping new SMS sends")
# Проверка статусов — всегда
delivered = step_check_status(conn, settings)
step_handle_expired(conn, settings)
log.info(f"Run summary: sent={sent}, delivered={delivered}")
if sent > 0 or delivered > 0:
send_telegram(
f"📦 <b>Платное хранение</b>\nОтправлено: {sent}\nДоставлено: {delivered}",
settings.get("telegram_chat_id", TELEGRAM_CHAT_ID),
)
# Обновляем баланс
if sent > 0:
api_id = settings.get("sms_api_id", SMS_API_ID)
balance, raw = fetch_balance(api_id)
if balance is not None:
with conn.cursor() as cur:
cur.execute(
"UPDATE sms_campaign_settings SET last_balance = %s WHERE campaign_type = 'paid_storage'",
(balance,)
)
conn.commit()
log.info(f"Balance updated: {balance} ₽")
except Exception as e:
log.error(f"Fatal error: {e}", exc_info=True)
finally:
conn.close()
log.info("Paid Storage Campaign — END")
log.info("=" * 60)
if __name__ == "__main__":
main()