107 lines
2.7 KiB
Python
107 lines
2.7 KiB
Python
#!/usr/bin/env python3
|
|
"""SQLite 任务持久化"""
|
|
import json
|
|
import sqlite3
|
|
import time
|
|
import uuid
|
|
|
|
from config import DB_PATH
|
|
|
|
|
|
def _conn():
|
|
c = sqlite3.connect(DB_PATH, timeout=30)
|
|
c.row_factory = sqlite3.Row
|
|
return c
|
|
|
|
|
|
def init_db():
|
|
c = _conn()
|
|
c.execute('''
|
|
CREATE TABLE IF NOT EXISTS tasks (
|
|
id TEXT PRIMARY KEY,
|
|
url TEXT NOT NULL,
|
|
goal TEXT NOT NULL,
|
|
status TEXT DEFAULT 'queued',
|
|
result TEXT DEFAULT 'pending',
|
|
max_steps INTEGER DEFAULT 30,
|
|
timeout INTEGER DEFAULT 600,
|
|
created_at REAL,
|
|
started_at REAL,
|
|
finished_at REAL,
|
|
steps INTEGER DEFAULT 0,
|
|
summary TEXT,
|
|
error TEXT,
|
|
report_path TEXT
|
|
)
|
|
''')
|
|
c.commit()
|
|
c.close()
|
|
|
|
|
|
def create_task(url, goal, max_steps, timeout):
|
|
tid = uuid.uuid4().hex[:12]
|
|
c = _conn()
|
|
c.execute(
|
|
'INSERT INTO tasks (id, url, goal, status, max_steps, timeout, created_at) '
|
|
'VALUES (?,?,?,?,?,?,?)',
|
|
(tid, url, goal, 'queued', max_steps, timeout, time.time()))
|
|
c.commit()
|
|
c.close()
|
|
return tid
|
|
|
|
|
|
def update_task(tid, **fields):
|
|
allowed = {'status', 'result', 'started_at', 'finished_at', 'steps',
|
|
'summary', 'error', 'report_path'}
|
|
sets = [f'{k}=?' for k in fields if k in allowed]
|
|
vals = [fields[k] for k in fields if k in allowed]
|
|
if not sets:
|
|
return
|
|
c = _conn()
|
|
c.execute(f'UPDATE tasks SET {", ".join(sets)} WHERE id=?', (*vals, tid))
|
|
c.commit()
|
|
c.close()
|
|
|
|
|
|
def get_task(tid):
|
|
c = _conn()
|
|
row = c.execute('SELECT * FROM tasks WHERE id=?', (tid,)).fetchone()
|
|
c.close()
|
|
return dict(row) if row else None
|
|
|
|
|
|
def list_tasks(limit=50):
|
|
c = _conn()
|
|
rows = c.execute(
|
|
'SELECT * FROM tasks ORDER BY created_at DESC LIMIT ?', (limit,)
|
|
).fetchall()
|
|
c.close()
|
|
return [dict(r) for r in rows]
|
|
|
|
|
|
def save_step_log(tid, step):
|
|
"""把单个步骤 JSON 追加到任务目录的 steps.jsonl"""
|
|
from config import TASKS_DIR
|
|
import os
|
|
p = os.path.join(TASKS_DIR, tid, 'steps.jsonl')
|
|
with open(p, 'a', encoding='utf-8') as f:
|
|
f.write(json.dumps(step, ensure_ascii=False) + '\n')
|
|
|
|
|
|
def load_step_logs(tid):
|
|
from config import TASKS_DIR
|
|
import os
|
|
p = os.path.join(TASKS_DIR, tid, 'steps.jsonl')
|
|
if not os.path.exists(p):
|
|
return []
|
|
out = []
|
|
with open(p, encoding='utf-8') as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if line:
|
|
try:
|
|
out.append(json.loads(line))
|
|
except json.JSONDecodeError:
|
|
pass
|
|
return out
|