diff --git a/scripts/autocall_scheduler.py b/scripts/autocall_scheduler.py new file mode 100644 index 0000000..073dc95 --- /dev/null +++ b/scripts/autocall_scheduler.py @@ -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() diff --git a/supabase/migrations/20260902_autocall_queue.sql b/supabase/migrations/20260902_autocall_queue.sql new file mode 100644 index 0000000..1844043 --- /dev/null +++ b/supabase/migrations/20260902_autocall_queue.sql @@ -0,0 +1,20 @@ +-- ============================================================ +-- SuperSam: Автодозвон — очередь ручных звонков (Фаза 3, п.9) +-- 02.09.2026. Аддитивная. Убирает нехватку autocall_queue из базовой миграции. +-- Применять: docker exec supabase-db psql -U supabase_admin -d postgres -f +-- ============================================================ + +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).';