feat(autocall): добавить autocall_queue (ручной звонок через планировщик). Миграция 20260902 + чтение очереди в scheduler
This commit is contained in:
Hermes
parent
75476c9d70
commit
4ed7f64a37
|
|
@ -0,0 +1,287 @@
|
||||||
|
#!/usr/bin/env python3
|
||||||
|
"""
|
||||||
|
SuperSam — Автодозвон: планировщик (Фаза 3)
|
||||||
|
Генерирует call-файлы для АТС клиента (Asterisk) и кладёт их в spool через SSH.
|
||||||
|
|
||||||
|
Архитектура: systemd timer (каждые N минут) + этот скрипт.
|
||||||
|
Каждый запуск:
|
||||||
|
1. Читает настройки кампании autocall_slot из sms_campaign_settings.
|
||||||
|
2. Находит группы, которым пора звонить (тайминг от sms_sent_at + пауза).
|
||||||
|
3. Генерирует call-файл (Context: SS-Autocall-Slot) и кладёт на АТС.
|
||||||
|
4. В test_mode звонит только на test_phone.
|
||||||
|
|
||||||
|
Безопасность: сценарий НЕ активируется сам — только если enabled=true в БД.
|
||||||
|
"""
|
||||||
|
import os
|
||||||
|
import sys
|
||||||
|
import time
|
||||||
|
import logging
|
||||||
|
import subprocess
|
||||||
|
from datetime import datetime, timezone, timedelta
|
||||||
|
|
||||||
|
import psycopg2
|
||||||
|
from psycopg2.extras import RealDictCursor
|
||||||
|
|
||||||
|
# ─── Конфигурация ────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
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")
|
||||||
|
|
||||||
|
# АТС (клиентская, через OpenVPN)
|
||||||
|
ATS_HOST = os.environ.get("ATS_HOST", "192.168.122.203")
|
||||||
|
ATS_USER = os.environ.get("ATS_USER", "root")
|
||||||
|
ATS_PASS = os.environ.get("ATS_PASS", "tigrissoro")
|
||||||
|
ATS_SPOOL = "/var/spool/asterisk/outgoing"
|
||||||
|
CALLER_ID = "2565551100"
|
||||||
|
|
||||||
|
# Edge function для приёма результата
|
||||||
|
API_URL = os.environ.get("AUTOCALL_API_URL", "https://supa.supersamsev.ru/functions/v1/autocall-result")
|
||||||
|
API_KEY = os.environ.get("AUTOCALL_API_KEY", "a7e1a7e9b543d22a93b29cb1291cbf7b3dd2ec67f1a4dae2")
|
||||||
|
|
||||||
|
LOG_FILE = "/var/log/supersam-autocall.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("autocall")
|
||||||
|
|
||||||
|
# ─── БД ──────────────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
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):
|
||||||
|
with conn.cursor(cursor_factory=RealDictCursor) as cur:
|
||||||
|
cur.execute("SELECT * FROM sms_campaign_settings WHERE campaign_type = 'autocall_slot'")
|
||||||
|
row = cur.fetchone()
|
||||||
|
if not row:
|
||||||
|
return {
|
||||||
|
"enabled": False, "test_mode": True, "test_phone": "9067029953",
|
||||||
|
"work_hours_start": 9, "work_hours_end": 20, "work_days": "1,2,3,4,5,6",
|
||||||
|
"delay_after_sms_minutes": 90, "retry_interval_minutes": 120,
|
||||||
|
"attempts_per_day": 3, "max_attempts": 5,
|
||||||
|
"no_input_escalate_after": 2, "next_day_attempt_times": "10:00,14:00",
|
||||||
|
"call_timeout_seconds": 30,
|
||||||
|
}
|
||||||
|
return dict(row)
|
||||||
|
|
||||||
|
def get_groups_to_call(conn, settings):
|
||||||
|
"""Группы, которым пора звонить."""
|
||||||
|
delay_min = settings.get("delay_after_sms_minutes", 90)
|
||||||
|
retry_min = settings.get("retry_interval_minutes", 120)
|
||||||
|
max_attempts = settings.get("max_attempts", 5)
|
||||||
|
no_input_escalate = settings.get("no_input_escalate_after", 2)
|
||||||
|
|
||||||
|
with conn.cursor(cursor_factory=RealDictCursor) as cur:
|
||||||
|
cur.execute("""
|
||||||
|
SELECT og.id, og.customer_phone, og.customer_phone_normalized,
|
||||||
|
og.notification_status, og.delivery_status, og.sms_sent_at,
|
||||||
|
og.autocall_opt_out,
|
||||||
|
(SELECT count(*) FROM autocall_log l
|
||||||
|
WHERE l.order_group_id = og.id) as attempt_count,
|
||||||
|
(SELECT count(*) FROM autocall_log l
|
||||||
|
WHERE l.order_group_id = og.id
|
||||||
|
AND l.outcome IN ('no_input_day','no_input_half')) as no_input_count,
|
||||||
|
(SELECT max(l.created_at) FROM autocall_log l
|
||||||
|
WHERE l.order_group_id = og.id) as last_call_at
|
||||||
|
FROM order_groups og
|
||||||
|
WHERE og.delivery_status = 'pending_confirmation'
|
||||||
|
AND og.autocall_opt_out = false
|
||||||
|
AND og.notification_status NOT IN ('confirmed', 'manual_required', 'opt_out')
|
||||||
|
AND og.sms_sent_at IS NOT NULL
|
||||||
|
-- Пауза после SMS прошла
|
||||||
|
AND og.sms_sent_at < NOW() - make_interval(mins => %s)
|
||||||
|
-- Не звонили или прошёл интервал повтора
|
||||||
|
AND (
|
||||||
|
NOT EXISTS (SELECT 1 FROM autocall_log l WHERE l.order_group_id = og.id)
|
||||||
|
OR (SELECT max(l.created_at) FROM autocall_log l
|
||||||
|
WHERE l.order_group_id = og.id) < NOW() - make_interval(mins => %s)
|
||||||
|
)
|
||||||
|
-- Не превышен лимит попыток
|
||||||
|
AND (SELECT count(*) FROM autocall_log l
|
||||||
|
WHERE l.order_group_id = og.id) < %s
|
||||||
|
ORDER BY og.sms_sent_at ASC
|
||||||
|
""", (delay_min, retry_min, max_attempts))
|
||||||
|
return [dict(r) for r in cur.fetchall()]
|
||||||
|
|
||||||
|
def is_work_time(settings, now=None):
|
||||||
|
"""Проверяет, входит ли сейчас в рабочее окно."""
|
||||||
|
now = now or datetime.now(timezone.utc)
|
||||||
|
# Рабочие часы в МСК (UTC+3)
|
||||||
|
msk = now.astimezone(timezone(timedelta(hours=3)))
|
||||||
|
hour = msk.hour
|
||||||
|
start = settings.get("work_hours_start", 9)
|
||||||
|
end = settings.get("work_hours_end", 20)
|
||||||
|
work_days = [int(d) for d in str(settings.get("work_days", "1,2,3,4,5,6")).split(",") if d.strip().isdigit()]
|
||||||
|
# Python: Monday=0, Sunday=6. БД: 1=Пн ... 7=Вс
|
||||||
|
weekday = msk.weekday() + 1
|
||||||
|
if weekday not in work_days:
|
||||||
|
return False
|
||||||
|
return start <= hour < end
|
||||||
|
|
||||||
|
def normalize_phone(phone):
|
||||||
|
"""Нормализует телефон: только цифры, начинается с 7."""
|
||||||
|
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 build_call_file(group, settings, attempt_no):
|
||||||
|
"""Формирует содержимое call-файла."""
|
||||||
|
phone = group.get("customer_phone_normalized") or group.get("customer_phone", "")
|
||||||
|
phone = normalize_phone(phone)
|
||||||
|
# Формат набора: 0 + номер без ведущей 7
|
||||||
|
dial = "0" + phone[1:] if phone.startswith("7") else "0" + phone
|
||||||
|
|
||||||
|
test_mode = settings.get("test_mode", True)
|
||||||
|
if test_mode:
|
||||||
|
test_phone = normalize_phone(settings.get("test_phone", "9067029953"))
|
||||||
|
dial = "0" + test_phone[1:] if test_phone.startswith("7") else "0" + test_phone
|
||||||
|
|
||||||
|
scenario = group.get("scenario", "A") # выбор дня + времени
|
||||||
|
call_type = "dlv"
|
||||||
|
|
||||||
|
lines = [
|
||||||
|
f"Channel: PJSIP/{dial}@mmedia",
|
||||||
|
f"CallerID: {CALLER_ID}",
|
||||||
|
"Context: SS-Autocall-Slot",
|
||||||
|
"Extension: s",
|
||||||
|
"Priority: 1",
|
||||||
|
"MaxRetries: 0",
|
||||||
|
"WaitTime: 45",
|
||||||
|
"Archive: yes",
|
||||||
|
f"Setvar: SCENARIO={scenario}",
|
||||||
|
f"Setvar: CALLTYPE={call_type}",
|
||||||
|
f"Setvar: GROUP_ID={group['id']}",
|
||||||
|
f"Setvar: ATTEMPT={attempt_no}",
|
||||||
|
f"Setvar: CUSTOMER_PHONE={phone}",
|
||||||
|
f"Setvar: API_URL={API_URL}",
|
||||||
|
f"Setvar: API_KEY={API_KEY}",
|
||||||
|
f"Setvar: TEST_MODE={1 if test_mode else 0}",
|
||||||
|
"",
|
||||||
|
]
|
||||||
|
return "\n".join(lines)
|
||||||
|
|
||||||
|
def push_call_file(group, settings, attempt_no):
|
||||||
|
"""Кладёт call-файл на АТС через SSH."""
|
||||||
|
content = build_call_file(group, settings, attempt_no)
|
||||||
|
filename = f"ss-{group['id'][:8]}-{int(time.time())}.call"
|
||||||
|
# Пишем файл на АТС через sshpass + cat
|
||||||
|
cmd = [
|
||||||
|
"sshpass", "-p", ATS_PASS,
|
||||||
|
"ssh", "-o", "StrictHostKeyChecking=no", "-o", "UserKnownHostsFile=/dev/null",
|
||||||
|
"-o", "ConnectTimeout=10", "-q",
|
||||||
|
f"{ATS_USER}@{ATS_HOST}",
|
||||||
|
f"cat > {ATS_SPOOL}/{filename}",
|
||||||
|
]
|
||||||
|
try:
|
||||||
|
proc = subprocess.run(cmd, input=content.encode(), capture_output=True, timeout=20)
|
||||||
|
if proc.returncode != 0:
|
||||||
|
log.error(f"Failed to push call file for {group['id']}: {proc.stderr.decode()[:200]}")
|
||||||
|
return False
|
||||||
|
log.info(f"Pushed call file {filename} for group {group['id']} (phone={group.get('customer_phone')})")
|
||||||
|
return True
|
||||||
|
except Exception as e:
|
||||||
|
log.error(f"SSH push error for {group['id']}: {e}")
|
||||||
|
return False
|
||||||
|
|
||||||
|
def get_queued_calls(conn):
|
||||||
|
"""Ручные запросы «Позвонить сейчас» из autocall_queue."""
|
||||||
|
with conn.cursor(cursor_factory=RealDictCursor) as cur:
|
||||||
|
cur.execute("""
|
||||||
|
SELECT q.id as queue_id, q.order_group_id, q.phone, q.scenario
|
||||||
|
FROM autocall_queue q
|
||||||
|
WHERE q.processed_at IS NULL
|
||||||
|
ORDER BY q.requested_at ASC
|
||||||
|
""")
|
||||||
|
rows = [dict(r) for r in cur.fetchall()]
|
||||||
|
# Подтянем имя/тип доставки из order_groups для call-файла
|
||||||
|
if rows:
|
||||||
|
ids = tuple(r["order_group_id"] for r in rows)
|
||||||
|
cur.execute("""
|
||||||
|
SELECT og.id, og.customer_phone, og.customer_phone_normalized
|
||||||
|
FROM order_groups og WHERE og.id IN %s
|
||||||
|
""", (ids,))
|
||||||
|
og = {r["id"]: r for r in cur.fetchall()}
|
||||||
|
for r in rows:
|
||||||
|
g = og.get(r["order_group_id"])
|
||||||
|
if g:
|
||||||
|
r["customer_phone_normalized"] = g.get("customer_phone_normalized")
|
||||||
|
r.setdefault("customer_phone", g.get("customer_phone"))
|
||||||
|
return rows
|
||||||
|
|
||||||
|
|
||||||
|
def mark_queue_processed(conn, queue_id):
|
||||||
|
with conn.cursor() as cur:
|
||||||
|
cur.execute("UPDATE autocall_queue SET processed_at = now() WHERE id = %s", (queue_id,))
|
||||||
|
conn.commit()
|
||||||
|
|
||||||
|
|
||||||
|
def process_queue(conn, settings):
|
||||||
|
"""Забирает ручные запросы и создаёт call-файлы. Возвращает число созданных."""
|
||||||
|
queued = get_queued_calls(conn)
|
||||||
|
if not queued:
|
||||||
|
return 0
|
||||||
|
# Номера попыток для ручных звонков — с учётом уже сделанных автозвонков
|
||||||
|
attempt_map = {}
|
||||||
|
pushed = 0
|
||||||
|
for q in queued:
|
||||||
|
group_id = q["order_group_id"]
|
||||||
|
if group_id not in attempt_map:
|
||||||
|
with conn.cursor() as cur:
|
||||||
|
cur.execute("SELECT count(*) FROM autocall_log WHERE order_group_id = %s", (group_id,))
|
||||||
|
attempt_map[group_id] = (cur.fetchone()[0] or 0) + 1
|
||||||
|
attempt_no = attempt_map[group_id]
|
||||||
|
g = dict(q)
|
||||||
|
g["id"] = group_id
|
||||||
|
if push_call_file(g, settings, attempt_no):
|
||||||
|
mark_queue_processed(conn, q["queue_id"])
|
||||||
|
pushed += 1
|
||||||
|
attempt_map[group_id] = attempt_no + 1
|
||||||
|
return pushed
|
||||||
|
|
||||||
|
|
||||||
|
def main():
|
||||||
|
conn = get_db_conn()
|
||||||
|
try:
|
||||||
|
settings = load_settings(conn)
|
||||||
|
|
||||||
|
# 0) Ручные звонки «Позвонить сейчас» — даже если кампания выключена,
|
||||||
|
# но в рабочее время и enabled_or_queue. Ручной запрос — явный сигнал.
|
||||||
|
if is_work_time(settings):
|
||||||
|
nq = process_queue(conn, settings)
|
||||||
|
if nq:
|
||||||
|
log.info(f"Processed {nq} queued manual calls")
|
||||||
|
else:
|
||||||
|
log.info("Outside work hours, manual queue not processed")
|
||||||
|
|
||||||
|
if not settings.get("enabled", False):
|
||||||
|
log.info("Autocall campaign disabled (enabled=false), skipping auto")
|
||||||
|
return
|
||||||
|
|
||||||
|
groups = get_groups_to_call(conn, settings)
|
||||||
|
log.info(f"Found {len(groups)} groups to call")
|
||||||
|
|
||||||
|
pushed = 0
|
||||||
|
for group in groups:
|
||||||
|
attempt_no = (group.get("attempt_count") or 0) + 1
|
||||||
|
if push_call_file(group, settings, attempt_no):
|
||||||
|
pushed += 1
|
||||||
|
time.sleep(1) # пауза между звонками
|
||||||
|
|
||||||
|
log.info(f"Pushed {pushed} call files")
|
||||||
|
finally:
|
||||||
|
conn.close()
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
main()
|
||||||
|
|
@ -0,0 +1,20 @@
|
||||||
|
-- ============================================================
|
||||||
|
-- SuperSam: Автодозвон — очередь ручных звонков (Фаза 3, п.9)
|
||||||
|
-- 02.09.2026. Аддитивная. Убирает нехватку autocall_queue из базовой миграции.
|
||||||
|
-- Применять: docker exec supabase-db psql -U supabase_admin -d postgres -f <file>
|
||||||
|
-- ============================================================
|
||||||
|
|
||||||
|
create table if not exists public.autocall_queue (
|
||||||
|
id bigint generated always as identity primary key,
|
||||||
|
order_group_id uuid not null references public.order_groups(id) on delete cascade,
|
||||||
|
phone text not null,
|
||||||
|
scenario text not null default 'A',
|
||||||
|
requested_by uuid,
|
||||||
|
requested_at timestamptz not null default now(),
|
||||||
|
processed_at timestamptz
|
||||||
|
);
|
||||||
|
create index if not exists idx_autocall_queue_pending
|
||||||
|
on public.autocall_queue(processed_at) where processed_at is null;
|
||||||
|
grant select, insert, update on public.autocall_queue to authenticated;
|
||||||
|
comment on table public.autocall_queue
|
||||||
|
is 'Ручные запросы «Позвонить сейчас». Планировщик забирает processed_at is null и создаёт call-файл (без ssh из edge fn).';
|
||||||
Loading…
Reference in New Issue