左辰勇
137 lines
5.1 KiB
Python
137 lines
5.1 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""导出 device_liveness 表里所有 device_id 及其对应的推送 regId,按 push_vendor 分组(本地开发工具)。
|
|
|
|
用法:
|
|
双击 scripts/show_device_regids.bat,或命令行:
|
|
python scripts/show_device_regids.py [filter]
|
|
|
|
[filter] 可选:大小写不敏感的子串,匹配 device_id / push_vendor / push_token /
|
|
registration_id 任一列;不传则列出全部。
|
|
show_device_regids.py xiaomi # 只看小米
|
|
show_device_regids.py device_Pixel # 按 device_id 片段找
|
|
|
|
说明:
|
|
- 直接以**只读**方式读 SQLite(server 在跑也不会抢写锁),所以开不开服务器都能用。
|
|
- push_token = 各厂商的 regId/pushToken(新链路,当前在用);registration_id = 旧极光
|
|
regId(历史兼容)。两列都打出来,方便跟 logcat 现役值逐字比对。
|
|
- 输出刻意保持纯 ASCII 排版并竖排展示**完整值**:双击弹出的 cmd 走 GBK 码页,竖排纯
|
|
ASCII 不会乱码;完整值不截断,才能直接复制去比对。
|
|
- 若 DATABASE_URL 不是 sqlite(生产 postgres),这里只提示改用 psql。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
|
|
|
|
def _resolve_sqlite_path() -> Path | None:
|
|
"""按 env > .env > 默认 的顺序拿 DATABASE_URL,解析出 sqlite 文件路径。"""
|
|
url = os.environ.get("DATABASE_URL", "").strip()
|
|
if not url:
|
|
env = REPO_ROOT / ".env"
|
|
if env.exists():
|
|
for line in env.read_text(encoding="utf-8", errors="ignore").splitlines():
|
|
s = line.strip()
|
|
if s.startswith("DATABASE_URL="):
|
|
url = s.split("=", 1)[1].strip().strip('"').strip("'")
|
|
break
|
|
if not url:
|
|
url = "sqlite:///./data/app.db"
|
|
if not url.startswith("sqlite"):
|
|
print(f"[!] DATABASE_URL not sqlite: {url}")
|
|
print(" This tool only reads a local SQLite dev DB. For prod use psql.")
|
|
return None
|
|
raw = url.split("///", 1)[1] if "///" in url else "./data/app.db"
|
|
p = Path(raw)
|
|
if not p.is_absolute():
|
|
p = (REPO_ROOT / raw).resolve()
|
|
return p
|
|
|
|
|
|
def main() -> int:
|
|
needle = sys.argv[1].lower() if len(sys.argv) > 1 else None
|
|
|
|
db = _resolve_sqlite_path()
|
|
if db is None:
|
|
return 2
|
|
if not db.exists():
|
|
print(f"[X] DB file not found: {db}")
|
|
return 2
|
|
|
|
# 只读打开,避免和正在运行的 server 抢写锁。URI 路径用正斜杠(as_posix)规避
|
|
# Windows 反斜杠/盘符在 file: URI 里的歧义;万一 URI 形式打不开再回退普通连接。
|
|
try:
|
|
con = sqlite3.connect(f"file:{db.as_posix()}?mode=ro", uri=True)
|
|
except sqlite3.OperationalError:
|
|
con = sqlite3.connect(str(db))
|
|
con.row_factory = sqlite3.Row
|
|
rows = con.execute(
|
|
"""
|
|
SELECT id, user_id, device_id, push_vendor, push_token, registration_id,
|
|
platform, app_version, liveness_state, last_heartbeat_at, updated_at
|
|
FROM device_liveness
|
|
ORDER BY updated_at DESC
|
|
"""
|
|
).fetchall()
|
|
con.close()
|
|
|
|
if needle:
|
|
def hit(r: sqlite3.Row) -> bool:
|
|
for k in ("device_id", "push_vendor", "push_token", "registration_id"):
|
|
v = r[k]
|
|
if v and needle in str(v).lower():
|
|
return True
|
|
return False
|
|
|
|
rows = [r for r in rows if hit(r)]
|
|
|
|
with_token = sum(1 for r in rows if (r["push_token"] or "").strip())
|
|
with_reg = sum(1 for r in rows if (r["registration_id"] or "").strip())
|
|
|
|
# 按 push_vendor 分组;None/空归入 "(none)"。组顺序:设备数多的在前,(none) 垫底;
|
|
# 组内沿用 updated_at DESC(rows 查询时已如此排序,dict 保序即可)。
|
|
groups: dict[str, list] = {}
|
|
for r in rows:
|
|
groups.setdefault(r["push_vendor"] or "(none)", []).append(r)
|
|
ordered = sorted(
|
|
groups.items(), key=lambda kv: (kv[0] == "(none)", -len(kv[1]), kv[0])
|
|
)
|
|
tally = " ".join(f"{v}={len(items)}" for v, items in ordered) or "-"
|
|
|
|
print(f"DB : {db}")
|
|
line = f"device_liveness : {len(rows)} row(s)"
|
|
if needle:
|
|
line += f' filter="{sys.argv[1]}"'
|
|
print(line)
|
|
print(f"has push_token(vendor regId) : {with_token}"
|
|
f" has registration_id(jiguang) : {with_reg}")
|
|
print(f"vendors : {tally}")
|
|
print("=" * 72)
|
|
|
|
if not rows:
|
|
print("(no rows)")
|
|
return 0
|
|
|
|
for vendor, items in ordered:
|
|
print()
|
|
print(f"===== vendor={vendor} : {len(items)} device(s) =====")
|
|
for r in items:
|
|
print(f" [#{r['id']}] user_id={r['user_id']} platform={r['platform']}"
|
|
f" state={r['liveness_state']} app={r['app_version'] or '-'}")
|
|
print(f" device_id : {r['device_id']}")
|
|
print(f" push_token(regId): {r['push_token'] or '(empty)'}")
|
|
print(f" registration_id : {r['registration_id'] or '(empty)'}")
|
|
print(f" last_heartbeat : {r['last_heartbeat_at'] or '-'}"
|
|
f" updated : {r['updated_at']}")
|
|
print(" " + "-" * 68)
|
|
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|