首次推送

2026-10-18 19:44:38 +08:00
commit f047662129
64 changed files with 10454 additions and 0 deletions
@@ -0,0 +1,94 @@
+"""测试辅助：直接读写测试库，用于构造 HTTP 层做不到的前置条件。
+
+用法：
+  python dbtool.py backdate-pause <db> <sub_id> <days>   把暂停开始时间提前 N 天
+  python dbtool.py bulk <db> <count>                     批量造客户+按日订阅（包月日常）
+  python dbtool.py exec <db> <sql>                       执行一条 SQL
+  python dbtool.py scalar <db> <sql>                     输出第一条结果的第一列
+  python dbtool.py integrity <db>                        完整性检查
+"""
+import sqlite3
+import sys
+import datetime
+
+
+def connect(path):
+    con = sqlite3.connect(path, timeout=10)
+    con.execute("PRAGMA busy_timeout=10000")
+    con.execute("PRAGMA foreign_keys=ON")
+    return con
+
+
+def backdate_pause(db, sub_id, days):
+    con = connect(db)
+    now = datetime.datetime.now() - datetime.timedelta(days=int(days))
+    stamp = now.strftime("%Y-%m-%d %H:%M:%S")
+    cur = con.execute(
+        "UPDATE sub_pauses SET start_at=? WHERE sub_id=? AND end_at IS NULL",
+        (stamp, int(sub_id)))
+    con.commit()
+    print("updated rows:", cur.rowcount, "start_at:", stamp)
+    con.close()
+
+
+def bulk(db, count):
+    con = connect(db)
+    today = datetime.date.today()
+    start = (today - datetime.timedelta(days=5)).isoformat()
+    end = (today + datetime.timedelta(days=25)).isoformat()
+    stamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+    pkg = con.execute("SELECT id FROM packages WHERE unit='day' ORDER BY id LIMIT 1").fetchone()
+    if not pkg:
+        print("FAIL: 没有按日套餐")
+        sys.exit(1)
+    pkg_id = pkg[0]
+    created = 0
+    for i in range(int(count)):
+        tail = "B%04d" % i
+        con.execute(
+            "INSERT OR IGNORE INTO customers(tail,name,created_at,updated_at) VALUES(?,?,?,?)",
+            (tail, "批量客户%d" % i, stamp, stamp))
+        row = con.execute("SELECT id FROM customers WHERE tail=?", (tail,)).fetchone()
+        cid = row[0]
+        con.execute(
+            "INSERT OR IGNORE INTO subscriptions(customer_id,package_id,unit,status,start_date,end_date,"
+            "total_count,used_count,created_at,updated_at) VALUES(?,?,'day','active',?,?,0,0,?,?)",
+            (cid, pkg_id, start, end, stamp, stamp))
+        created += 1
+    con.commit()
+    total = con.execute("SELECT COUNT(*) FROM subscriptions").fetchone()[0]
+    print("bulk customers:", created, "subscriptions total:", total)
+    con.close()
+
+
+def main():
+    if len(sys.argv) < 3:
+        print(__doc__)
+        sys.exit(2)
+    cmd = sys.argv[1]
+    db = sys.argv[2]
+    if cmd == "backdate-pause":
+        backdate_pause(db, sys.argv[3], sys.argv[4])
+    elif cmd == "bulk":
+        bulk(db, sys.argv[3])
+    elif cmd == "exec":
+        con = connect(db)
+        cur = con.execute(sys.argv[3])
+        con.commit()
+        print("rows:", cur.rowcount)
+        con.close()
+    elif cmd == "scalar":
+        con = connect(db)
+        print(con.execute(sys.argv[3]).fetchone()[0])
+        con.close()
+    elif cmd == "integrity":
+        con = connect(db)
+        print(con.execute("PRAGMA integrity_check").fetchone()[0])
+        con.close()
+    else:
+        print("未知命令:", cmd)
+        sys.exit(2)
+
+
+if __name__ == "__main__":
+    main()