95 lines
3.2 KiB
Python
95 lines
3.2 KiB
Python
"""测试辅助：直接读写测试库，用于构造 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()
|