chongming/chongming-license-gui
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

modify: 台账更改插件

Browse Source
This commit is contained in:
chakcy
2026-09-22 16:08:12 +08:00
parent f468845f67
commit e1c02a9b80
37 changed files with 6377 additions and 380 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/pg_check.py
+279
View File
@@ -0,0 +1,279 @@
"""对**真实 PostgreSQL 服务器**做台账契约一致性校验。
为什么需要它：本工具开发机上没有任何 PostgreSQL（无服务、无 Docker），因此
PG 后端的真实读写语义无法在开发环境里验证。这个脚本让你**在自己的库上**
一次性验完，而且：
* **全程在一个事务里，最后 ``ROLLBACK``** —— 连建表都在事务内，
对你的真实库**零残留**（表本来存在就什么都不变）；
* 用**同一套契约**分别跑本地 SQLite 与 PostgreSQL，然后**逐项对比**，
这样"两个后端行为不一致"会直接暴露成 FAIL，而不是等同事用出问题才发现；
* 写入的记录 ``lic_id`` 一律带 ``PGCHECK-`` 前缀，且**所有观测值都只统计这些
记录** —— 库里已有的真实台账既不参与对比，也不会被打印出来；
* 不会改动你的任何真实数据（验证结束后事务回滚）。
用法::
# 1) 显式参数（密码走环境变量或交互输入，避免留在命令历史里）
uv run python tools/pg_check.py --host 127.0.0.1 --port 5432 \\
--dbname chongming_licenses --user license_ro
# 2) 或用 DSN
uv run python tools/pg_check.py --dsn "postgresql://license_ro@127.0.0.1:5432/chongming_licenses"
# 密码来源：环境变量 PGPASSWORD 或交互式提问（推荐）；
# 也支持 --password（会留在命令历史里，仅用于自动化）
退出码 0 = 两个后端行为一致。
"""
from __future__ import annotations
import argparse
import getpass
import os
import sys
import tempfile
from datetime import timedelta
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT / "src"))
from chongming_license import ( # noqa: E402
Registry,
build_license,
build_payload,
generate_keypair,
machine_code,
utc_today,
write_license,
)
from chongming_license_gui import pg_store # noqa: E402
from chongming_license_gui.pg_store import ( # noqa: E402
DEFAULT_DBNAME,
DEFAULT_HOST,
DEFAULT_PORT,
DEFAULT_SCHEMA,
DEFAULT_SSLMODE,
SSL_MODES,
PgConfig,
PostgresRegistry,
)
PREFIX = "PGCHECK-"
STATUSES = ("expiring", "forever", "active", "expired", "not_yet_valid")
failures: list[str] = []
def check(name: str, ok: bool, detail: str = "") -> None:
print(f"[{'PASS' if ok else 'FAIL'}] {name}" + (f" — {detail}" if detail else ""))
if not ok:
failures.append(name)
def _mine(records) -> list[str]:
"""只保留本次校验写入的记录（真实台账不参与对比、也不被打印）。"""
return [r.lic_id for r in records if r.lic_id.startswith(PREFIX)]
# ---------------------------------------------------------------------------
# 契约：在任意后端上跑同一套操作，返回"可比较的观测值"
# ---------------------------------------------------------------------------
def _demo_docs(sandbox: Path) -> list[dict]:
"""两条用于校验的授权：一条即将到期、一条永久（覆盖排序哨兵与状态分桶）。
客户名刻意混入 ASCII（``PGCheck``），用来验证"大小写不敏感"这个方言差异点。
"""
private_key, _ = generate_keypair()
today = utc_today()
sources = {
"board_serial": "pgcheckboard01",
"product_uuid": "pgcheckproductuuid0123456789ab",
"mac": "001b44pgcheck",
"machine_id": "pgcheckmachineid0123456789abcdef",
}
specs = [
(f"{PREFIX}EXPIRING", "PGCheck 契约校验（示例）", ["aibox_sophon_adapter"],
(today + timedelta(days=20)).isoformat()),
(f"{PREFIX}FOREVER", "PGCheck 永久（示例）",
["aibox_sophon_adapter", "ip_sound_column"], None),
]
docs = []
for lic_id, customer, plugins, expires in specs:
payload = build_payload(
machine_code=machine_code(sources), plugins=plugins, sources=sources,
customer=customer, lic_id=lic_id, expires_at=expires, note="载荷备注",
)
doc = build_license(payload, private_key, kid="pg-check")
write_license(doc, sandbox / f"{lic_id}.lic")
docs.append(doc)
return docs
def contract(reg, docs: list[dict], sandbox: Path) -> dict:
"""同一套操作 → 观测值。两边结果必须逐项相同。"""
ids = [d["payload"]["lic_id"] for d in docs]
mine = lambda records: _mine(records) # noqa: E731 - 短小，读起来更直观
obs: dict[str, object] = {}
for doc in docs:
reg.save(doc, public_key_fp="PGCHECK-FP", source="pg-check")
# 排序：到期日升序，永久（NULL）用哨兵排最后
obs["顺序(到期升序/永久最后)"] = mine(reg.query())
obs["客户模糊检索"] = mine(reg.query(customer="永久"))
obs["客户检索大小写不敏感"] = mine(reg.query(customer="pgcheck"))
obs["插件精确检索"] = sorted(mine(reg.query(plugin="ip_sound_column")))
obs["关键字检索"] = sorted(mine(reg.query(keyword="契约校验")))
obs["状态筛选(永久)"] = mine(reg.query(status="forever"))
obs["状态分桶"] = {s: len(mine(reg.query(status=s))) for s in STATUSES}
rec = reg.get(ids[0])
obs["get 命中"] = rec.lic_id if rec else None
obs["机器码已大写"] = bool(rec) and rec.machine_code == rec.machine_code.upper()
obs["来源指纹回读一致"] = bool(rec) and rec.sources == docs[0]["payload"]["sources"]
# 人工批注：手工改过之后，重新入库不得被载荷备注冲掉
obs["update_note 返回值"] = reg.update_note(ids[0], "人工批注")
reg.save(docs[0], public_key_fp="PGCHECK-FP", source="renew")
obs["重新入库后仍是人工批注"] = reg.get(ids[0]).note
# 首次入库时间要保留（库的 SQLite 版把 created_at 排除在 DO UPDATE 之外）
created_before = reg.get(ids[0]).created_at
reg.save(docs[0], public_key_fp="PGCHECK-FP", source="renew")
obs["重新入库后 created_at 不变"] = reg.get(ids[0]).created_at == created_before
# 导出：文件名规则、CSV 表头与编码（这些方法是从库继承的，两边应完全一致）
my_records = [r for r in reg.query() if r.lic_id.startswith(PREFIX)]
out = sandbox / "export"
obs["导出 .lic 文件名"] = sorted(p.name for p in reg.export_licenses(out, my_records))
csv_path = reg.export_csv(sandbox / "台账.csv", my_records)
raw = csv_path.read_bytes()
obs["CSV 带 BOM"] = raw[:3] == b"\xef\xbb\xbf"
lines = csv_path.read_text(encoding="utf-8-sig").splitlines()
obs["CSV 行数(含表头)"] = len(lines)
obs["CSV 表头列数"] = len(lines[0].split(",")) if lines else 0
obs["delete 返回值"] = reg.delete(ids[1])
obs["删除后剩余"] = mine(reg.query())
return obs
# ---------------------------------------------------------------------------
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="在真实 PostgreSQL 上校验台账后端与本地 SQLite 行为一致",
)
parser.add_argument("--dsn",
help="连接串（postgresql://用户@主机:端口/库名）；给了它就不用下面各项")
parser.add_argument("--host", default=DEFAULT_HOST)
parser.add_argument("--port", type=int, default=DEFAULT_PORT, help=f"默认 {DEFAULT_PORT}")
parser.add_argument("--dbname", default=DEFAULT_DBNAME)
parser.add_argument("--user", default="")
parser.add_argument("--password", default=None,
help="不建议：会留在命令历史里。优先用 PGPASSWORD 或交互输入")
parser.add_argument("--sslmode", default=DEFAULT_SSLMODE, choices=list(SSL_MODES))
parser.add_argument("--schema", default=DEFAULT_SCHEMA)
return parser.parse_args()
def resolve_config(args: argparse.Namespace) -> tuple[PgConfig, str]:
"""返回 (配置, 密码)。密码优先级：--password > DSN 内 > PGPASSWORD。"""
if args.dsn:
# 连接串解析交给 pg_store（进而交给 psycopg），避免自己写一套连接串文法
config = PgConfig.from_dsn(args.dsn, schema=args.schema)
password = config.password
else:
config = PgConfig(host=args.host, port=args.port, dbname=args.dbname,
user=args.user, sslmode=args.sslmode, schema=args.schema)
password = ""
if args.password:
password = args.password
return config, password or os.environ.get("PGPASSWORD", "")
def main() -> int:
args = parse_args()
try:
config, password = resolve_config(args)
except ImportError:
print("未安装 PostgreSQL 驱动：uv sync --extra postgres")
return 2
if not config.dbname:
print("请指定数据库名（--dbname 或 --dsn）")
return 2
if not password:
password = getpass.getpass(f"{config.user or 'postgres'}@{config.host} 的密码（不显示）：")
print("=" * 70)
print("台账契约一致性校验：本地 SQLite vs PostgreSQL")
print(f"目标：{config.display()}") # 展示串不含密码
print("说明：所有写入都在一个事务里，结束时 ROLLBACK；")
print(" 观测值只统计本次的 PGCHECK-* 记录，你的真实台账不参与、也不会被打印。")
print("=" * 70)
sandbox = Path(tempfile.mkdtemp(prefix="clg-pgcheck-"))
docs = _demo_docs(sandbox)
# ---- 1) 本地 SQLite 基准 ----
print("\n--- 本地 SQLite（基准）---")
sqlite_sandbox = sandbox / "sqlite"
sqlite_sandbox.mkdir(parents=True, exist_ok=True)
with Registry(sqlite_sandbox / "ledger.db") as sqlite_reg:
sqlite_obs = contract(sqlite_reg, docs, sqlite_sandbox)
for key, value in sqlite_obs.items():
print(f" {key} = {value}")
# ---- 2) PostgreSQL（一个事务，最后回滚）----
print("\n--- PostgreSQL（事务内执行，结束回滚）---")
pg_sandbox = sandbox / "pg"
pg_sandbox.mkdir(parents=True, exist_ok=True)
try:
reg = PostgresRegistry(config, password=password, autocommit=False)
except Exception as exc: # noqa: BLE001
print(f"[FAIL] 连接失败：{exc}")
return 1
try:
pg_obs = contract(reg, docs, pg_sandbox)
except Exception as exc: # noqa: BLE001 - 把差异当成校验失败而不是崩溃
print(f"[FAIL] 在 PostgreSQL 上执行契约时出错：{type(exc).__name__}: {exc}")
reg._conn.rollback()
reg.close()
return 1
# ---- 3) 逐项对比 ----
print("\n--- 逐项对比（SQLite 基准 vs PostgreSQL）---")
for key, expected in sqlite_obs.items():
actual = pg_obs.get(key)
check(key, actual == expected, "" if actual == expected
else f"SQLite={expected!r} PG={actual!r}")
# ---- 4) 回滚并确认没有残留 ----
reg._conn.rollback()
table = f'"{config.schema}"."{pg_store.TABLE}"'
try:
row = reg._conn.execute(
f"SELECT count(*) AS n FROM {table} WHERE lic_id LIKE '{PREFIX}%'"
).fetchone()
residue = row["n"] if row else 0
print(f"\n回滚完成：库里残留的校验记录数 = {residue}")
if residue:
print("[FAIL] 残留了校验记录 —— 这不该发生")
failures.append("回滚后有残留")
except Exception as exc: # noqa: BLE001
# 表本来不存在 → 连建表一起被回滚了，这正是期望结果
print(f"\n回滚完成：表不存在（建表也随事务一起回滚）—— {type(exc).__name__}")
reg.close()
print("\n" + "=" * 70)
if failures:
print(f"❌ {len(failures)} 项在两个后端之间不一致：{failures}")
print(" 请把上面的差异贴给研发 —— 这正是本脚本存在的意义。")
return 1
print("✅ PostgreSQL 后端与本地 SQLite 行为完全一致（且已在事务中回滚，无残留）")
return 0
if __name__ == "__main__":
sys.exit(main())
tools/screenshot.py
+235
View File
@@ -0,0 +1,235 @@
"""生成文档用的界面截图（无头运行）。
用法::
uv run python tools/screenshot.py
产物写入 ``docs/images/``，供 ``docs/使用手册.md`` 引用；改完主题重跑即可。
**安全约定（重要）**：本脚本只画界面，绝不接触真实资产 ——
* ``QSettings`` 被重定向到临时 INI，不会读写用户真实的配置；
* 台账库指向临时目录，不会碰 ``~/.chongming/licenses.db``；
* 私钥是当场生成的一次性密钥，截图里不会出现任何真实密钥或客户信息；
* 演示数据统一带「示例」字样，一眼可与真实台账区分。
"""
from __future__ import annotations
import hashlib
import os
import sys
import tempfile
from datetime import timedelta
from pathlib import Path
# 必须在导入/创建 QApplication 之前设置：无头渲染，且不弹任何窗口
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT / "src"))
from PySide6.QtWidgets import QApplication # noqa: E402
OUT_DIR = ROOT / "docs" / "images"
WINDOW_WIDTH = 1240
def _demo_sources(seed: str) -> dict[str, str]:
"""按种子造一台"设备"的来源指纹（已按库规则归一化：小写、仅 [0-9a-z]）。"""
digest = hashlib.sha256(seed.encode("utf-8")).hexdigest()
return {
"board_serial": f"cnx{digest[:12]}",
"product_uuid": digest[12:44],
"mac": f"001b44{digest[44:56]}",
"machine_id": digest[56:88],
}
def main() -> int:
# ---- 1) 隔离：配置、台账、密钥全部落在临时目录 ----------------------
sandbox = Path(tempfile.mkdtemp(prefix="license-gui-shots-"))
app = QApplication.instance() or QApplication([])
from chongming_license import (
build_license,
build_payload,
encode_request_code,
generate_keypair,
machine_code,
utc_today,
write_license,
write_private_key,
)
from chongming_license_gui import common, theme
from chongming_license_gui.app import MainWindow
from chongming_license_gui.registry_tab import ChangePluginsDialog, RenewDialog
theme.apply_theme(app)
# 一次性密钥（截图里出现的是这把，与任何真实密钥无关）
private_key, public_b64 = generate_keypair()
key_path = sandbox / "demo_private.pem"
write_private_key(private_key, key_path, overwrite=True)
# 配置隔离到临时目录；并用 fileName() 自证，而不是"以为隔离了"
settings = common.isolated_settings(sandbox)
common.assert_isolated_settings(settings, sandbox)
window = MainWindow(settings)
window.signing_key.load(str(key_path))
fingerprint = window.signing_key.fingerprint
# ---- 2) 演示台账：五种状态各一条，正好展示状态配色 ------------------
registry_tab = window.registry_tab
registry_tab.db_edit.setText(str(sandbox / "demo-licenses.db"))
registry_tab._open_db(silent=True)
today = utc_today()
demo = [
# (客户, 种子, 插件, not_before 偏移, expires 偏移, 台账批注)
("华东水务集团（示例）", "dev-water-01", ["aibox_sophon_adapter"],
None, -26, "合同 SM-2025-0417 · 已到期待续"),
("上海崇明智慧农业（示例）", "dev-farm-07", ["aibox_sophon_adapter", "ip_sound_column"],
None, 12, "已电话确认续期，等客户回申请码"),
("宁波港机（示例）", "dev-port-03", ["aibox_sophon_adapter"],
None, 248, "现场调试中"),
("苏州工业园区（示例）", "dev-suzhou-11", ["ip_sound_column"],
None, None, "战略合作 · 永久授权"),
("舟山渔场（示例）", "dev-zhoushan-05", ["aibox_sophon_adapter"],
45, 410, "新设备，发货前预签发"),
]
def _day(offset: int | None) -> str | None:
return None if offset is None else (today + timedelta(days=offset)).isoformat()
for customer, seed, plugins, not_before, expires, note in demo:
sources = _demo_sources(seed)
payload = build_payload(
machine_code=machine_code(sources),
plugins=plugins,
sources=sources,
customer=customer,
not_before=_day(not_before),
expires_at=_day(expires),
note="示例数据（由 tools/screenshot.py 生成）",
)
doc = build_license(payload, private_key, kid="vendor-demo")
window.registry.save(doc, public_key_fp=fingerprint, source="gui")
registry_tab.refresh()
# ---- 3) 签发页填好一份"正在签发"的数据 ------------------------------
issue = window.issue_tab
request_code = encode_request_code(_demo_sources("dev-new-42"))
issue.rc_edit.setPlainText(request_code)
issue._parse_request_code()
issue._set_all_checked(True)
issue.customer_edit.setText("南通冷链物流（示例）")
issue.note_edit.setText("合同 NT-2026-0921")
issue.kid_edit.setText("vendor-2026")
issue.days_spin.setValue(365)
issue._refresh_preview()
# ---- 4) 查看/校验页：先出一张"通过"，再出一张"异机被拒" -----------
# 用**有效**（而非已过期）的那份来演示"校验通过"，否则截图会是一张红条：
# 过期件也"签名有效"，但结论是失败，拿它当"通过"示例会误导阅读手册的人。
ok_index = 2 # 宁波港机（示例）：active
ok_sources = _demo_sources(demo[ok_index][1])
ok_license = sandbox / "license_active.lic"
ok_payload = build_payload(
machine_code=machine_code(ok_sources),
plugins=demo[ok_index][2],
sources=ok_sources,
customer=demo[ok_index][0],
expires_at=_day(demo[ok_index][4]),
note="示例数据（由 tools/screenshot.py 生成）",
)
write_license(build_license(ok_payload, private_key, kid="vendor-demo"), ok_license)
inspect = window.inspect_tab
inspect.lic_edit.setText(str(ok_license))
inspect.pub_edit.setText(public_b64)
inspect.rb_remote.setChecked(True)
inspect.rc_edit.setText(encode_request_code(ok_sources))
inspect._run()
# ---- 5) 逐页截图 ---------------------------------------------------
OUT_DIR.mkdir(parents=True, exist_ok=True)
shots: list[Path] = []
def _grab(widget, name: str) -> Path:
widget.show()
# 两轮事件循环：第一轮完成布局，第二轮把重绘落到位，再抓图才不会是半成品
QApplication.processEvents()
QApplication.processEvents()
target = OUT_DIR / name
widget.grab().save(str(target), "PNG")
shots.append(target)
return target
def _fit(page, minimum: int = 700) -> int:
"""把窗口调到"本页内容刚好放得下"，避免截图中出现滚动条或截断。"""
area = window._area_by_page[page]
height = minimum
while height < 2000:
window.resize(WINDOW_WIDTH, height)
QApplication.processEvents()
if area.verticalScrollBar().maximum() == 0:
break
height += 60
return height
# 签发页
window.tabs.setCurrentWidget(window._area_by_page[issue])
_fit(issue)
_grab(window, "01-issue.png")
# 台账页
window.tabs.setCurrentWidget(window._area_by_page[registry_tab])
_fit(registry_tab)
_grab(window, "02-registry.png")
# 密钥页
window.tabs.setCurrentWidget(window._area_by_page[window.keys_tab])
_fit(window.keys_tab)
_grab(window, "03-keys.png")
# 查看/校验页：通过
window.tabs.setCurrentWidget(window._area_by_page[inspect])
_fit(inspect)
_grab(window, "04-inspect-ok.png")
# 查看/校验页：异机被拒（换一台机器的申请码）
inspect.rc_edit.setText(encode_request_code(_demo_sources("dev-other-99")))
inspect._run()
_grab(window, "05-inspect-fail.png")
# 续期对话框
records = registry_tab.records
renew_target = next((r for r in records if r.status() == "expiring"), records[0])
dialog = RenewDialog(renew_target, window)
dialog.resize(660, dialog.sizeHint().height())
_grab(dialog, "06-renew.png")
dialog.close()
# 改插件对话框：勾一个原先没有的、去掉一个原先有的，"新增/移除"差异才看得见
plugins_dialog = ChangePluginsDialog(renew_target, window.settings, window)
plugins_dialog.plugin_picker.ensure_types(["video_matrix_adapter"])
plugins_dialog.plugin_picker.set_checked(
[*renew_target.plugins[:1], "video_matrix_adapter"]
)
plugins_dialog.resize(660, plugins_dialog.sizeHint().height())
_grab(plugins_dialog, "07-change-plugins.png")
plugins_dialog.close()
# ---- 6) 收尾 -------------------------------------------------------
print(f"截图输出目录：{OUT_DIR}")
for path in shots:
print(f" {path.name} {path.stat().st_size // 1024} KB")
print(f"\n演示用临时目录（可直接删除）：{sandbox}")
return 0
if __name__ == "__main__":
sys.exit(main())