94 lines
3.6 KiB
Python
94 lines
3.6 KiB
Python
"""diag: 查 scheduled-e621c8a6 这个 job 为何执行到一半没推送(ASCII only, GBK safe)."""
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
env = Path(__file__).resolve().parent.parent / ".env"
|
|
for line in env.read_text(encoding="utf-8").splitlines():
|
|
if line.strip().startswith("ZCBOT_DB_URL="):
|
|
os.environ["ZCBOT_DB_URL"] = line.split("=", 1)[1].strip()
|
|
from sqlalchemy import create_engine, text # noqa: E402
|
|
import builtins # noqa: E402
|
|
|
|
_out = open(Path(__file__).resolve().parent / "_sched_e621.txt", "w", encoding="utf-8")
|
|
|
|
|
|
def print(*a, **k): # noqa: A001
|
|
builtins.print(*a, **k, file=_out)
|
|
|
|
|
|
PREFIX = sys.argv[1] if len(sys.argv) > 1 else "e621c8a6"
|
|
engine = create_engine(os.environ["ZCBOT_DB_URL"])
|
|
|
|
|
|
def s(x, n=2000):
|
|
t = str(x if x is not None else "")
|
|
return t if len(t) <= n else t[:n] + f"...[+{len(t)-n}]"
|
|
|
|
|
|
with engine.connect() as conn:
|
|
job = conn.execute(text(
|
|
"select job_id,user_id,name,mode,cron,tz,enabled,notify,timeout_seconds,"
|
|
"next_run_at,last_run_at,last_status,last_error,last_task_id,"
|
|
"consecutive_failures,run_count,bound_task_id,created_at,deleted_at "
|
|
"from scheduled_jobs where cast(job_id as text) like :p"),
|
|
{"p": PREFIX + "%"}).fetchall()
|
|
print(f"[JOBS matched '{PREFIX}'] {len(job)}")
|
|
for j in job:
|
|
print("-" * 60)
|
|
print(f"job_id={j[0]} name={j[2]!r}")
|
|
print(f" mode={j[3]} cron={j[4]!r} tz={j[5]} enabled={j[6]} timeout={j[8]}")
|
|
print(f" notify={j[7]}")
|
|
print(f" next_run_at={j[9]} last_run_at={j[10]}")
|
|
print(f" last_status={j[11]} consecutive_failures={j[14]} run_count={j[15]}")
|
|
print(f" last_task_id={j[13]} bound_task_id={j[16]}")
|
|
print(f" deleted_at={j[18]} created_at={j[17]}")
|
|
if j[12]:
|
|
print(f" last_error: {s(j[12], 1500)}")
|
|
|
|
if not job:
|
|
sys.exit(0)
|
|
|
|
j = job[0]
|
|
uid = j[1]
|
|
last_tid = j[13]
|
|
|
|
# 找该 job 关联的所有 task(scheduled_job_id 回填 + last_task_id)
|
|
tasks = conn.execute(text(
|
|
"select task_id,name,status,run_status,run_error,tokens_prompt,tokens_completion,"
|
|
"created_at,updated_at,scheduled_job_id from tasks "
|
|
"where scheduled_job_id = :jid order by created_at"),
|
|
{"jid": str(j[0])}).fetchall()
|
|
print("\n" + "=" * 60)
|
|
print(f"[TASKS with scheduled_job_id={str(j[0])[:8]}] {len(tasks)}")
|
|
for t in tasks:
|
|
print(f" task={t[0]} name={t[1]!r} status={t[2]} run={t[3]} "
|
|
f"tok={t[5]}/{t[6]} created={t[7]} updated={t[8]}")
|
|
if t[4]:
|
|
print(f" run_error: {s(t[4], 1500)}")
|
|
|
|
# dump last_task_id 的消息(执行到哪一步)
|
|
tid = last_tid or (tasks[-1][0] if tasks else None)
|
|
if tid is None:
|
|
print("\n[no task to dump]")
|
|
sys.exit(0)
|
|
print("\n" + "=" * 60)
|
|
print(f"[DUMP messages of task {tid}]")
|
|
msgs = conn.execute(text(
|
|
"select idx,payload,tokens_in,tokens_out,created_at from messages "
|
|
"where task_id=:t order by idx"), {"t": str(tid)}).fetchall()
|
|
print(f"messages: {len(msgs)}\n")
|
|
for idx, p, ti, to, cat in msgs:
|
|
role = p.get("role")
|
|
head = f"[{idx}] {role} tok={ti}/{to} at={cat}"
|
|
print(head)
|
|
content = p.get("content")
|
|
if content:
|
|
print(" content:", s(content, 1500))
|
|
for tc in p.get("tool_calls") or []:
|
|
fn = tc.get("function") or {}
|
|
print(f" CALL {fn.get('name')}({s(fn.get('arguments'), 800)})")
|
|
if role == "tool":
|
|
print(f" TOOL[{p.get('name')}]:", s(content, 1200))
|
|
print()
|