Files
dsh_ai1net_server/归档/接续历史/机制排查/S12_一次性排期盘点_20261002.py
admin c1b5e4d966 chore(工作区): 全量入库 + 补齐 .gitignore(以工作区为准)
- 变更规模:新增 514 / 修改 62 / 重命名 155 / 删除 4(归档重组与文档轮次)
- .gitignore 修:`归档/**/db-cwd归一-备份-*/` —— 原规则写绝对层级(归档/db-cwd归一-…),
  目录搬进 归档/配置与备份/ 后**静默失效**,43 MB 的 DB 备份又变成未跟踪
- .gitignore 补:嵌套 git 内部数据(归档/内嵌git-20261008/、归档/skills-git-旧线-20261007/dotgit-原样移出/)
- .gitignore 补:运行态与部署副本(.workbuddy/collab/、.workbuddy/tools/、.workbuddy/.load-pending、.workbuddy/tmp-*)
- .gitignore 补:备份件(*.bak-*)
- 未跟踪文件从 2190 降到 890(其余为 归档/ 归档件与 .workbuddy/memory/ 知识文件,按口径入库)
2026-10-10 23:13:22 +08:00

35 lines
1.6 KiB
Python

import sqlite3, sys, json
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
DB = r"E:/ProgramData/.workbuddy/workbuddy.db"
c = sqlite3.connect(DB, timeout=15)
c.execute("PRAGMA busy_timeout=8000")
TARGET = ("deleted_at IS NULL AND status='ACTIVE' AND schedule_type='once' "
"AND next_run_at IS NULL")
n = c.execute(f"select count(*) from automations where {TARGET}").fetchone()[0]
print("BEFORE_ONCE_ACTIVE_NO_NEXT =", n)
print("TOTAL_ROWS =", c.execute("select count(*) from automations").fetchone()[0])
print("ACTIVE_ALL =", c.execute(
"select count(*) from automations where deleted_at is null and status='ACTIVE'"
).fetchone()[0])
print("RECURRING_ACTIVE =", c.execute(
"select count(*) from automations where deleted_at is null and status='ACTIVE' "
"and schedule_type='recurring'").fetchone()[0])
print("--- group: schedule_type x status x next_run_is_null ---")
for r in c.execute(
"select schedule_type, status, (next_run_at is null), count(*) from automations "
"where deleted_at is null group by 1,2,3 order by 1,2,3"):
print(" ", r)
print("--- target rows (full) ---")
cols = [d[1] for d in c.execute("PRAGMA table_info(automations)")]
print("COLUMNS =", cols)
for r in c.execute(f"select * from automations where {TARGET}"):
d = dict(zip(cols, r))
print(" ", json.dumps({k: d.get(k) for k in
("id", "name", "status", "schedule_type", "next_run_at", "created_at",
"updated_at", "deleted_at")}, ensure_ascii=False))
print("INTEGRITY_BEFORE =", c.execute("PRAGMA integrity_check").fetchone()[0])
c.close()