181 lines
7.5 KiB
Go
181 lines
7.5 KiB
Go
package storage
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"time"
|
||
|
||
"github.com/jackc/pgx/v5"
|
||
)
|
||
|
||
// PBXCall — звонок АТС для записи в базу.
|
||
type PBXCall struct {
|
||
UID, Type, Status, Client, ClientKey, Via, ViaKey, UserLogin, UserName, RecordURL string
|
||
StartedAt time.Time
|
||
Wait, Duration int
|
||
Raw json.RawMessage
|
||
}
|
||
|
||
// UpsertPBXCalls сохраняет звонки одним пакетом; повторная загрузка того же периода обновляет записи.
|
||
func (s *Storage) UpsertPBXCalls(ctx context.Context, clinic string, calls []PBXCall) error {
|
||
if len(calls) == 0 {
|
||
return nil
|
||
}
|
||
b := &pgx.Batch{}
|
||
for _, c := range calls {
|
||
raw := c.Raw
|
||
if len(raw) == 0 {
|
||
raw = json.RawMessage(`{}`)
|
||
}
|
||
b.Queue(`
|
||
INSERT INTO pbx_calls (uid, type, status, client, client_key, via, via_key, user_login, user_name,
|
||
started_at, wait, duration, record_url, raw, clinic, synced_at)
|
||
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15, now())
|
||
ON CONFLICT (clinic, uid) DO UPDATE SET type=EXCLUDED.type, status=EXCLUDED.status, client=EXCLUDED.client,
|
||
client_key=EXCLUDED.client_key, via=EXCLUDED.via, via_key=EXCLUDED.via_key,
|
||
user_login=EXCLUDED.user_login, user_name=EXCLUDED.user_name, started_at=EXCLUDED.started_at,
|
||
wait=EXCLUDED.wait, duration=EXCLUDED.duration, record_url=EXCLUDED.record_url,
|
||
raw=EXCLUDED.raw, synced_at=now()`,
|
||
c.UID, c.Type, c.Status, c.Client, c.ClientKey, c.Via, c.ViaKey, c.UserLogin, c.UserName,
|
||
c.StartedAt, c.Wait, c.Duration, c.RecordURL, raw, clinic)
|
||
}
|
||
return s.pool.SendBatch(ctx, b).Close()
|
||
}
|
||
|
||
func (s *Storage) StartPBXSync(ctx context.Context, clinic string, from, to time.Time, manual bool) (int64, error) {
|
||
var id int64
|
||
err := s.pool.QueryRow(ctx, `INSERT INTO pbx_sync_log (period_from, period_to, manual, clinic) VALUES ($1,$2,$3,$4) RETURNING id`,
|
||
from, to, manual, clinic).Scan(&id)
|
||
return id, err
|
||
}
|
||
|
||
func (s *Storage) FinishPBXSync(ctx context.Context, id int64, calls int, errText string) error {
|
||
_, err := s.pool.Exec(ctx, `UPDATE pbx_sync_log SET finished_at=now(), calls=$2, error=$3 WHERE id=$1`, id, calls, errText)
|
||
return err
|
||
}
|
||
|
||
type PBXStatus struct {
|
||
Calls int `json:"calls"`
|
||
Oldest *time.Time `json:"oldest"`
|
||
Newest *time.Time `json:"newest"`
|
||
LastSyncAt *time.Time `json:"last_sync_at"`
|
||
LastSyncOK *time.Time `json:"last_sync_ok"`
|
||
LastError string `json:"last_error"`
|
||
LastCalls int `json:"last_calls"`
|
||
LinesMapped int `json:"lines_mapped"`
|
||
LinesUnknown int `json:"lines_unknown"`
|
||
}
|
||
|
||
func (s *Storage) PBXStatus(ctx context.Context, clinic string) (*PBXStatus, error) {
|
||
st := &PBXStatus{}
|
||
if err := s.pool.QueryRow(ctx, `SELECT count(*), min(started_at), max(started_at) FROM pbx_calls WHERE clinic=$1`, clinic).
|
||
Scan(&st.Calls, &st.Oldest, &st.Newest); err != nil {
|
||
return nil, err
|
||
}
|
||
err := s.pool.QueryRow(ctx, `
|
||
SELECT started_at, error, calls FROM pbx_sync_log WHERE clinic=$1 AND finished_at IS NOT NULL ORDER BY id DESC LIMIT 1`, clinic).
|
||
Scan(&st.LastSyncAt, &st.LastError, &st.LastCalls)
|
||
if err != nil && err != pgx.ErrNoRows {
|
||
return nil, err
|
||
}
|
||
_ = s.pool.QueryRow(ctx, `SELECT max(finished_at) FROM pbx_sync_log WHERE clinic=$1 AND error = '' AND finished_at IS NOT NULL`, clinic).Scan(&st.LastSyncOK)
|
||
if err := s.pool.QueryRow(ctx, `
|
||
SELECT count(*) FILTER (WHERE l.via_key IS NOT NULL), count(*) FILTER (WHERE l.via_key IS NULL)
|
||
FROM (SELECT DISTINCT via_key FROM pbx_calls WHERE clinic=$1 AND via_key <> '') v
|
||
LEFT JOIN pbx_lines l ON l.clinic = $1 AND l.via_key = v.via_key`, clinic).Scan(&st.LinesMapped, &st.LinesUnknown); err != nil {
|
||
return nil, err
|
||
}
|
||
return st, nil
|
||
}
|
||
|
||
// PBXLine — номер клиники, на который звонят, с каналом и статистикой.
|
||
type PBXLine struct {
|
||
ViaKey string `json:"via_key"`
|
||
Via string `json:"via"`
|
||
Channel string `json:"channel"`
|
||
Calls int `json:"calls"`
|
||
UserNames []string `json:"user_names"` // как называются сотрудники АТС на этой линии
|
||
LastCall *time.Time `json:"last_call"`
|
||
}
|
||
|
||
func (s *Storage) PBXLines(ctx context.Context, clinic string) ([]PBXLine, error) {
|
||
rows, err := s.pool.Query(ctx, `
|
||
SELECT k.via_key, COALESCE(c.via, ''), COALESCE(l.channel, ''), COALESCE(c.n, 0),
|
||
COALESCE(c.names, '{}'), c.last
|
||
FROM (SELECT via_key FROM pbx_lines WHERE clinic=$1
|
||
UNION SELECT DISTINCT via_key FROM pbx_calls WHERE clinic=$1 AND via_key <> '') k
|
||
LEFT JOIN pbx_lines l ON l.clinic = $1 AND l.via_key = k.via_key
|
||
LEFT JOIN LATERAL (
|
||
SELECT max(via) AS via, count(*) AS n, max(started_at) AS last,
|
||
(array_agg(DISTINCT user_name) FILTER (WHERE user_name <> ''))[1:4] AS names
|
||
FROM pbx_calls WHERE clinic = $1 AND via_key = k.via_key
|
||
) c ON true
|
||
ORDER BY COALESCE(c.n, 0) DESC, k.via_key`, clinic)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
out := []PBXLine{}
|
||
for rows.Next() {
|
||
var l PBXLine
|
||
if err := rows.Scan(&l.ViaKey, &l.Via, &l.Channel, &l.Calls, &l.UserNames, &l.LastCall); err != nil {
|
||
return nil, err
|
||
}
|
||
out = append(out, l)
|
||
}
|
||
return out, rows.Err()
|
||
}
|
||
|
||
// SetPBXLine задаёт канал номера; пустой канал снимает привязку.
|
||
func (s *Storage) SetPBXLine(ctx context.Context, clinic, viaKey, channel string) error {
|
||
if channel == "" {
|
||
_, err := s.pool.Exec(ctx, `DELETE FROM pbx_lines WHERE clinic=$1 AND via_key=$2`, clinic, viaKey)
|
||
return err
|
||
}
|
||
_, err := s.pool.Exec(ctx, `
|
||
INSERT INTO pbx_lines (clinic, via_key, channel) VALUES ($1,$2,$3)
|
||
ON CONFLICT (clinic, via_key) DO UPDATE SET channel=EXCLUDED.channel, updated_at=now()`, clinic, viaKey, channel)
|
||
return err
|
||
}
|
||
|
||
// PBXCallRow — звонок за период для отчётов.
|
||
type PBXCallRow struct {
|
||
UID string `json:"uid"`
|
||
Type string `json:"type"`
|
||
Status string `json:"status"`
|
||
Client string `json:"client"`
|
||
ClientKey string `json:"client_key"`
|
||
Via string `json:"via"`
|
||
ViaKey string `json:"via_key"`
|
||
Channel string `json:"channel"`
|
||
UserName string `json:"user_name"`
|
||
StartedAt time.Time `json:"started_at"`
|
||
Wait int `json:"wait"`
|
||
Duration int `json:"duration"`
|
||
RecordURL string `json:"record_url"`
|
||
}
|
||
|
||
// PBXCalls — звонки клиники за период [from, to) по времени, с каналом номера «Через».
|
||
func (s *Storage) PBXCalls(ctx context.Context, clinic string, from, to time.Time) ([]PBXCallRow, error) {
|
||
rows, err := s.pool.Query(ctx, `
|
||
SELECT c.uid, c.type, c.status, c.client, c.client_key, c.via, c.via_key, COALESCE(l.channel, ''),
|
||
c.user_name, c.started_at, c.wait, c.duration, c.record_url
|
||
FROM pbx_calls c LEFT JOIN pbx_lines l ON l.clinic = c.clinic AND l.via_key = c.via_key
|
||
WHERE c.clinic = $1 AND c.started_at >= $2 AND c.started_at < $3
|
||
ORDER BY c.started_at`, clinic, from, to)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
out := []PBXCallRow{}
|
||
for rows.Next() {
|
||
var r PBXCallRow
|
||
if err := rows.Scan(&r.UID, &r.Type, &r.Status, &r.Client, &r.ClientKey, &r.Via, &r.ViaKey, &r.Channel,
|
||
&r.UserName, &r.StartedAt, &r.Wait, &r.Duration, &r.RecordURL); err != nil {
|
||
return nil, err
|
||
}
|
||
out = append(out, r)
|
||
}
|
||
return out, rows.Err()
|
||
}
|