diff --git a/deploy/popadmin/_deploy_im_messages.py b/deploy/popadmin/_deploy_im_messages.py new file mode 100644 index 0000000..2831bfd --- /dev/null +++ b/deploy/popadmin/_deploy_im_messages.py @@ -0,0 +1,201 @@ +#!/usr/bin/env python3 +"""Deploy popadmin IM messages UI (timestamp + date range) to static-override. + +Also optionally rebuilds spark-server for start/end query support. +""" +from __future__ import annotations + +import os +import sys +import tarfile +import tempfile +import time +from pathlib import Path + +import paramiko + +ROOT = Path(__file__).resolve().parents[2] +HOST = "47.236.66.9" +USER = "root" +PEM = Path(os.path.expandvars(r"%USERPROFILE%\.ssh\wyrs_shanghai_deploy.pem")) +LOCAL_JS = ROOT / "deploy" / "popadmin" / "popadmin.js" +REMOTE_STATIC = "/opt/wyrs-admin/static-override" +CACHE = time.strftime("20261102-imts-%H%M%S") +REMOTE_BUILD = "/tmp/spark-im-messages-build" +INCLUDE = ["cmd", "internal", "migrations", "api", "go.mod", "go.sum"] +EXTRA_FILES = [ + (ROOT / "deploy" / "docker" / "Dockerfile.remote", "Dockerfile.remote"), + (ROOT / "deploy" / "docker" / "deploy_remote.sh", "deploy_remote.sh"), +] + + +def load_key(path: Path): + for loader in ( + paramiko.Ed25519Key.from_private_key_file, + paramiko.RSAKey.from_private_key_file, + paramiko.ECDSAKey.from_private_key_file, + ): + try: + return loader(str(path)) + except Exception: + continue + raise SystemExit(f"cannot load private key {path}") + + +def run(client: paramiko.SSHClient, cmd: str, timeout: int = 180) -> tuple[int, str]: + print("$", cmd[:240]) + _, stdout, stderr = client.exec_command(cmd, timeout=timeout) + out = stdout.read().decode("utf-8", "replace") + err = stderr.read().decode("utf-8", "replace") + code = stdout.channel.recv_exit_status() + text = (out + (("\n" + err) if err.strip() else "")).strip() + if text: + print(text[-12000:]) + return code, text + + +def deploy_static(client: paramiko.SSHClient, sftp: paramiko.SFTPClient) -> None: + ts = time.strftime("%Y%m%d%H%M%S") + print("=== backup + upload static ===") + run( + client, + f"cp -a {REMOTE_STATIC}/popadmin.js {REMOTE_STATIC}/popadmin.js.bak.imts.{ts} && " + f"cp -a {REMOTE_STATIC}/popadmin.index.html {REMOTE_STATIC}/popadmin.index.html.bak.imts.{ts}", + ) + sftp.put(str(LOCAL_JS), f"{REMOTE_STATIC}/popadmin.js") + idx_cmd = f"""python3 - <<'PY' +from pathlib import Path +import re +p = Path('{REMOTE_STATIC}/popadmin.index.html') +s = p.read_text(encoding='utf-8') +s2 = re.sub(r'popadmin\\.js\\?v=[^\"\\']+', 'popadmin.js?v={CACHE}', s) +p.write_text(s2, encoding='utf-8') +print('cache', '{CACHE}') +print('js_ok', 'popadmin.js?v={CACHE}' in s2) +print('has_fmtMsgTs', 'fmtMsgTs' in Path('{REMOTE_STATIC}/popadmin.js').read_text(encoding='utf-8', errors='replace')) +print('has_imPeriod', 'imPeriod' in Path('{REMOTE_STATIC}/popadmin.js').read_text(encoding='utf-8', errors='replace')) +PY""" + code, _ = run(client, idx_cmd) + if code != 0: + raise SystemExit(f"cache bump failed: {code}") + + +def make_tarball(path: Path) -> None: + with tarfile.open(path, "w:gz") as tar: + for name in INCLUDE: + p = ROOT / name + if not p.exists(): + raise SystemExit(f"missing {p}") + tar.add(p, arcname=name) + for src, arc in EXTRA_FILES: + if not src.is_file(): + raise SystemExit(f"missing {src}") + tar.add(src, arcname=arc) + print("tarball", path, path.stat().st_size, "bytes") + + +def deploy_backend(client: paramiko.SSHClient, sftp: paramiko.SFTPClient) -> None: + release = "im-range-" + time.strftime("%Y%m%d%H%M%S") + print("=== conflict probe ===") + code, out = run( + client, + "for f in /tmp/spark-image-limit-build.lock /tmp/spark-compose-mutate.lock " + "/tmp/sign-energy-deploy.lock /tmp/spark-im-messages-build.lock; do " + " if [ -f \"$f\" ]; then echo LOCKED:$f; cat \"$f\"; exit 99; fi; " + "done; echo locks_clear", + timeout=60, + ) + if code != 0: + raise SystemExit(f"deploy locked: {out}") + + with tempfile.TemporaryDirectory() as td: + tar_path = Path(td) / "spark-src.tgz" + make_tarball(tar_path) + run(client, f"rm -rf {REMOTE_BUILD}; mkdir -p {REMOTE_BUILD}", timeout=60) + print("upload tarball...") + sftp.put(str(tar_path), f"{REMOTE_BUILD}/spark-src.tgz") + for src, arc in EXTRA_FILES: + data = src.read_bytes().replace(b"\r\n", b"\n") + with sftp.file(f"{REMOTE_BUILD}/{arc}", "wb") as f: + f.write(data) + try: + f.chmod(0o755 if arc.endswith(".sh") else 0o644) + except Exception: + pass + + print("=== remote docker build ===") + code, out = run( + client, + f"cd {REMOTE_BUILD} && chmod +x deploy_remote.sh && " + f"SPARK_RELEASE={release} SPARK_SRC_DIR={REMOTE_BUILD} " + f"SPARK_DEPLOY_LOCK=/tmp/spark-im-messages-build.lock " + f"./deploy_remote.sh", + timeout=1800, + ) + if code != 0: + raise SystemExit(f"backend deploy failed: {code}") + # probe new string in binary + run( + client, + "docker exec spark-server sh -c " + "\"strings /usr/local/bin/spark-server | grep -F '查询跨度最长' | head -3\" " + "|| true", + timeout=60, + ) + + +def verify(client: paramiko.SSHClient) -> None: + print("=== verify ===") + run( + client, + "python3 - <<'PY'\n" + "from pathlib import Path\n" + f"js=Path('{REMOTE_STATIC}/popadmin.js').read_text(encoding='utf-8',errors='replace')\n" + f"idx=Path('{REMOTE_STATIC}/popadmin.index.html').read_text(encoding='utf-8',errors='replace')\n" + "print('fmtMsgTs', 'fmtMsgTs' in js)\n" + "print('imPeriod', 'imPeriod' in js)\n" + "print('exceedsMonths', 'exceedsMonths' in js)\n" + "print('start_end_q', 'start=' in js and 'imPeriod' in js)\n" + f"print('cache_line', [ln for ln in idx.splitlines() if 'popadmin.js' in ln][:2])\n" + "PY", + timeout=60, + ) + # API auth gate still 401; bad range after auth would 400 — check route alive + run( + client, + "curl -sS -o /tmp/im_msg.out -w '%{http_code}' " + "'http://127.0.0.1:18080/api/admin/spark/im/messages?limit=1&start=2020-01-01&end=2021-02-01'; " + "echo; head -c 200 /tmp/im_msg.out; echo", + timeout=30, + ) + + +def main() -> int: + if not PEM.is_file(): + print("missing pem", PEM) + return 1 + if not LOCAL_JS.is_file(): + print("missing", LOCAL_JS) + return 1 + do_backend = "--skip-backend" not in sys.argv + + client = paramiko.SSHClient() + client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) + client.connect(HOST, username=USER, pkey=load_key(PEM), timeout=30) + sftp = client.open_sftp() + try: + deploy_static(client, sftp) + if do_backend: + deploy_backend(client, sftp) + else: + print("skip backend") + verify(client) + print("DEPLOY DONE cache=", CACHE) + finally: + sftp.close() + client.close() + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/deploy/popadmin/popadmin.js b/deploy/popadmin/popadmin.js index 9e4011f..c11ee45 100644 --- a/deploy/popadmin/popadmin.js +++ b/deploy/popadmin/popadmin.js @@ -243,6 +243,54 @@ function esc(s) { +/** IM / 消息时间:支持秒或毫秒 Unix,输出本地可读时间。 */ + +function fmtMsgTs(ts) { + + const n = Number(ts); + + if (!Number.isFinite(n) || n <= 0) return '—'; + + const d = new Date(n < 1e12 ? n * 1000 : n); + + if (Number.isNaN(d.getTime())) return String(ts ?? '—'); + + const p = (x) => String(x).padStart(2, '0'); + + return `${d.getFullYear()}-${p(d.getMonth() + 1)}-${p(d.getDate())} ${p(d.getHours())}:${p(d.getMinutes())}:${p(d.getSeconds())}`; + +} + + + +/** 日期间隔是否超过 12 个自然月(含起止日)。 */ + +function exceedsMonths(startYmd, endYmd, months) { + + if (!startYmd || !endYmd) return false; + + const [ys, ms, ds] = startYmd.split('-').map(Number); + + const [ye, me, de] = endYmd.split('-').map(Number); + + const from = new Date(ys, ms - 1, ds); + + const end = new Date(ye, me - 1, de); + + if (end < from) return true; + + const lim = new Date(ys, ms - 1 + months, ds); + + // 含止日:止日不得晚于 start+months 的前一天 + + lim.setDate(lim.getDate() - 1); + + return end > lim; + +} + + + function renderLookupList(data, key) { const users = asList(data.users); @@ -4087,9 +4135,21 @@ const pages = { const stats = await api('/spark/im/stats').catch(() => ({})); + const now = new Date(); + + const y = now.getFullYear(); + + const m = String(now.getMonth() + 1).padStart(2, '0'); + + const d = String(now.getDate()).padStart(2, '0'); + + const today = `${y}-${m}-${d}`; + + const dateStyle = 'padding:8px;border-radius:8px;border:1px solid var(--border);background:#0b1220;color:var(--text)'; + $('content').innerHTML = ` -
MQTT IM 消息(spark-im-messages.json)· 含文本/名片/撤回标记 · 接口 GET /api/admin/spark/im/messages
+
IM 消息(V3 库)· 时间字段 ts(发送/入库)· 筛选跨度最长 12 个月 · GET /api/admin/spark/im/messages?start&end
@@ -4103,14 +4163,44 @@ const pages = {
-
+
- + + + + + + +
+
+ @@ -4119,11 +4209,119 @@ const pages = {
时间类型fromto/群内容撤回
`; - const load = async (q) => { + const ymd = (dt) => { + + const p = (x) => String(x).padStart(2, '0'); + + return `${dt.getFullYear()}-${p(dt.getMonth() + 1)}-${p(dt.getDate())}`; + + }; + + const syncPeriodUI = () => { + + const period = $('imPeriod').value; + + const custom = period === 'custom'; + + const all = period === 'all'; + + $('imCustomWrap').style.display = custom ? 'inline-flex' : 'none'; + + $('imDay').style.display = custom || all ? 'none' : ''; + + }; + + const rangeOf = () => { + + const period = $('imPeriod').value; + + if (period === 'all') return { start: '', end: '' }; + + if (period === 'custom') { + + return { start: $('imStart').value || today, end: $('imEnd').value || today }; + + } + + const day = $('imDay').value || today; + + const [yy, mm, dd] = day.split('-').map(Number); + + const anchor = new Date(yy, mm - 1, dd); + + if (period === 'day') return { start: day, end: day }; + + if (period === 'week') { + + const dow = (anchor.getDay() + 6) % 7; // Mon=0 + + const mon = new Date(anchor); + + mon.setDate(anchor.getDate() - dow); + + const sun = new Date(mon); + + sun.setDate(mon.getDate() + 6); + + return { start: ymd(mon), end: ymd(sun) }; + + } + + if (period === 'month') { + + const last = new Date(yy, mm, 0).getDate(); + + return { start: `${yy}-${String(mm).padStart(2, '0')}-01`, end: `${yy}-${String(mm).padStart(2, '0')}-${String(last).padStart(2, '0')}` }; + + } + + // year + + return { start: `${yy}-01-01`, end: `${yy}-12-31` }; + + }; + + const load = async () => { + + const q = ($('imQ').value || '').trim(); + + const { start, end } = rangeOf(); + + if (start && end) { + + if (end < start) { + + alert('结束日期不能早于开始日期'); + + return; + + } + + if (exceedsMonths(start, end, 12)) { + + alert('查询跨度最长 12 个月,请缩小时间段'); + + return; + + } + + } + + $('imRangeHint').textContent = start && end ? `当前区间:${start} ~ ${end}` : '当前区间:全部'; try { - const data = await api('/spark/im/messages?limit=100' + (q ? '&q=' + encodeURIComponent(q) : '')); + let url = '/spark/im/messages?limit=100'; + + if (q) url += '&q=' + encodeURIComponent(q); + + if (start && end) { + + url += '&start=' + encodeURIComponent(start) + '&end=' + encodeURIComponent(end); + + } + + const data = await api(url); const rows = asList(data.messages || data); @@ -4131,7 +4329,7 @@ const pages = { ? rows.map(m => ` - ${esc(m.ts)}${esc(m.type || 'text')} + ${esc(fmtMsgTs(m.ts))}${esc(m.type || 'text')} ${esc(m.from)}${esc(m.groupId || m.to)} @@ -4151,9 +4349,13 @@ const pages = { }; - await load(''); + $('imPeriod').onchange = syncPeriodUI; - $('imSearch').onclick = () => load(($('imQ').value || '').trim()); + syncPeriodUI(); + + await load(); + + $('imSearch').onclick = () => load(); }, diff --git a/internal/module/admin/logic/im.go b/internal/module/admin/logic/im.go index 7647e3f..eaccabd 100644 --- a/internal/module/admin/logic/im.go +++ b/internal/module/admin/logic/im.go @@ -6,8 +6,12 @@ import ( "strconv" "strings" "time" + + "spark-server/internal/pkg/errs" ) +const maxMessageRangeMonths = 12 + func (s *Service) ImStats(ctx context.Context) (*ImStats, error) { if !s.ready() { return nil, ErrNotReady @@ -15,15 +19,19 @@ func (s *Service) ImStats(ctx context.Context) (*ImStats, error) { return s.repo.ImStats(ctx) } -func (s *Service) ListMessages(ctx context.Context, q string, limit int) ([]ImMessageView, error) { +func (s *Service) ListMessages(ctx context.Context, q string, limit int, start, end string) ([]ImMessageView, error) { if !s.ready() { return nil, ErrNotReady } if limit <= 0 || limit > 500 { limit = 100 } + from, to, err := parseMessageTimeRange(start, end) + if err != nil { + return nil, err + } q = strings.TrimSpace(q) - rows, err := s.repo.ListRecentMessages(ctx, limit*3) + rows, err := s.repo.ListRecentMessages(ctx, limit*3, from, to) if err != nil { return nil, err } @@ -41,6 +49,40 @@ func (s *Service) ListMessages(ctx context.Context, q string, limit int) ([]ImMe return out, nil } +// parseMessageTimeRange accepts YYYY-MM-DD start/end (Asia/Shanghai, inclusive calendar days). +// Empty both means no filter; one-sided or span > 12 months is invalid. +func parseMessageTimeRange(start, end string) (from, to *time.Time, err error) { + start = strings.TrimSpace(start) + end = strings.TrimSpace(end) + if start == "" && end == "" { + return nil, nil, nil + } + if start == "" || end == "" { + return nil, nil, errs.ErrInvalidArgument.WithMsg("start 与 end 需同时提供(YYYY-MM-DD)") + } + loc, locErr := time.LoadLocation("Asia/Shanghai") + if locErr != nil { + loc = time.FixedZone("CST", 8*3600) + } + fromDay, err := time.ParseInLocation("2006-01-02", start, loc) + if err != nil { + return nil, nil, errs.ErrInvalidArgument.WithMsg("start 日期格式应为 YYYY-MM-DD") + } + endDay, err := time.ParseInLocation("2006-01-02", end, loc) + if err != nil { + return nil, nil, errs.ErrInvalidArgument.WithMsg("end 日期格式应为 YYYY-MM-DD") + } + if endDay.Before(fromDay) { + return nil, nil, errs.ErrInvalidArgument.WithMsg("end 不能早于 start") + } + // Inclusive end day → exclusive upper bound at next midnight. + toExclusive := endDay.AddDate(0, 0, 1) + if toExclusive.After(fromDay.AddDate(0, maxMessageRangeMonths, 0)) { + return nil, nil, errs.ErrInvalidArgument.WithMsg(fmt.Sprintf("查询跨度最长 %d 个月", maxMessageRangeMonths)) + } + return &fromDay, &toExclusive, nil +} + func (s *Service) ListConversations(ctx context.Context, userID int64) (convs []ImConversationView, groups []GroupView, stats *ImStats, err error) { if !s.ready() { return nil, nil, nil, ErrNotReady diff --git a/internal/module/admin/logic/service.go b/internal/module/admin/logic/service.go index d648f04..7e963e4 100644 --- a/internal/module/admin/logic/service.go +++ b/internal/module/admin/logic/service.go @@ -4,6 +4,7 @@ package logic import ( "context" "errors" + "time" adminentity "spark-server/internal/module/admin/entity" adminmodel "spark-server/internal/module/admin/model" @@ -51,7 +52,7 @@ type Repository interface { ListComplaints(ctx context.Context, limit int) ([]adminmodel.ComplaintRow, error) InsertComplaint(ctx context.Context, r *friendentity.Report) error ImStats(ctx context.Context) (*adminentity.ImStats, error) - ListRecentMessages(ctx context.Context, fetchLimit int) ([]adminmodel.MessageRow, error) + ListRecentMessages(ctx context.Context, fetchLimit int, from, to *time.Time) ([]adminmodel.MessageRow, error) ListUserConversations(ctx context.Context, userID int64) ([]adminmodel.ConversationRow, error) } diff --git a/internal/module/admin/logic/service_test.go b/internal/module/admin/logic/service_test.go index ff3c796..f5626e1 100644 --- a/internal/module/admin/logic/service_test.go +++ b/internal/module/admin/logic/service_test.go @@ -184,8 +184,22 @@ func (m *memoryAdmin) ImStats(context.Context) (*adminentity.ImStats, error) { return m.stats, nil } -func (m *memoryAdmin) ListRecentMessages(_ context.Context, _ int) ([]adminmodel.MessageRow, error) { - return m.msgs, nil +func (m *memoryAdmin) ListRecentMessages(_ context.Context, _ int, from, to *time.Time) ([]adminmodel.MessageRow, error) { + if from == nil && to == nil { + return m.msgs, nil + } + out := make([]adminmodel.MessageRow, 0, len(m.msgs)) + for i := range m.msgs { + t := m.msgs[i].CreatedAt + if from != nil && t.Before(*from) { + continue + } + if to != nil && !t.Before(*to) { + continue + } + out = append(out, m.msgs[i]) + } + return out, nil } func (m *memoryAdmin) ListUserConversations(_ context.Context, _ int64) ([]adminmodel.ConversationRow, error) { @@ -295,10 +309,21 @@ func TestAdminServiceOps(t *testing.T) { if err != nil || stats.MessageCount != 2 { t.Fatalf("stats %+v %v", stats, err) } - msgs, err := svc.ListMessages(ctx, "hi", 0) + msgs, err := svc.ListMessages(ctx, "hi", 0, "", "") if err != nil || len(msgs) != 1 { t.Fatalf("msgs %+v %v", msgs, err) } + day := now.In(time.FixedZone("CST", 8*3600)).Format("2006-01-02") + msgs, err = svc.ListMessages(ctx, "", 0, day, day) + if err != nil || len(msgs) != 1 { + t.Fatalf("msgs range %+v %v", msgs, err) + } + if _, err := svc.ListMessages(ctx, "", 0, "2020-01-01", "2021-02-01"); !errors.Is(err, errs.ErrInvalidArgument) { + t.Fatalf("want span reject got %v", err) + } + if _, err := svc.ListMessages(ctx, "", 0, "2024-01-01", ""); !errors.Is(err, errs.ErrInvalidArgument) { + t.Fatalf("want one-sided reject got %v", err) + } convs, gviews, st, err := svc.ListConversations(ctx, 11) if err != nil || st != nil || len(convs) != 2 || len(gviews) != 1 { t.Fatalf("convs %+v %+v %v", convs, gviews, err) diff --git a/internal/module/admin/model/im.go b/internal/module/admin/model/im.go index 4e2be26..e7b18b6 100644 --- a/internal/module/admin/model/im.go +++ b/internal/module/admin/model/im.go @@ -26,7 +26,7 @@ func (m *Repo) ImStats(ctx context.Context) (*adminentity.ImStats, error) { return out, nil } -func (m *Repo) ListRecentMessages(ctx context.Context, fetchLimit int) ([]MessageRow, error) { +func (m *Repo) ListRecentMessages(ctx context.Context, fetchLimit int, from, to *time.Time) ([]MessageRow, error) { db, err := m.session(ctx) if err != nil { return nil, err @@ -50,8 +50,19 @@ FROM t_group_message` CreatedAt time.Time `gorm:"column:created_at"` ChatType string `gorm:"column:chat_type"` } + where := " WHERE 1=1" + args := make([]any, 0, 3) + if from != nil { + where += " AND created_at >= ?" + args = append(args, *from) + } + if to != nil { + where += " AND created_at < ?" + args = append(args, *to) + } + args = append(args, fetchLimit) var rows []scanRow - err = db.Raw(`SELECT * FROM (`+sql+`) AS m ORDER BY created_at DESC LIMIT ?`, fetchLimit).Scan(&rows).Error + err = db.Raw(`SELECT * FROM (`+sql+`) AS m`+where+` ORDER BY created_at DESC LIMIT ?`, args...).Scan(&rows).Error if err != nil { return nil, mapDBErr(err) } diff --git a/internal/transport/http/handler/admin/im_groups.go b/internal/transport/http/handler/admin/im_groups.go index fd06c3a..1729dc7 100644 --- a/internal/transport/http/handler/admin/im_groups.go +++ b/internal/transport/http/handler/admin/im_groups.go @@ -50,8 +50,16 @@ func (h *Handler) ImMessages(c *gin.Context) { return } limit, _ := strconv.Atoi(c.DefaultQuery("limit", "100")) - list, err := h.svc.ListMessages(c.Request.Context(), c.Query("q"), limit) + list, err := h.svc.ListMessages(c.Request.Context(), c.Query("q"), limit, c.Query("start"), c.Query("end")) if err != nil { + if errors.Is(err, errs.ErrInvalidArgument) { + msg := "start/end 无效" + if ae, ok := errs.AsAppError(err); ok && ae.CustomMsg() { + msg = ae.Msg + } + Fail(c, http.StatusBadRequest, msg) + return + } Fail(c, http.StatusInternalServerError, err.Error()) return }