hz4th_coder
8f0efe28b6
- 新增处理会话表(process_sessions)和步骤表(process_steps) - 支持实时查看每个产品的处理步骤进度 - 支持暂停/继续/停止正在进行的处理 - 处理步骤包括：搜索内容库、搜索互联网、抓取网页、提取数据、填充字段、提交审核 - 新增处理监控页面 /process - 每个步骤记录详细数据、耗时、状态 - 历史记录可查看每个步骤的执行详情
834 lines
34 KiB
Python
834 lines
34 KiB
Python
"""
|
||
数据库模型和操作
|
||
"""
|
||
import sqlite3
|
||
import json
|
||
from datetime import datetime
|
||
from contextlib import contextmanager
|
||
from config import Config
|
||
|
||
class Database:
|
||
def __init__(self, db_path=None):
|
||
self.db_path = db_path or Config.DATABASE
|
||
self.init_db()
|
||
|
||
@contextmanager
|
||
def get_connection(self):
|
||
"""获取数据库连接"""
|
||
conn = sqlite3.connect(self.db_path)
|
||
conn.row_factory = sqlite3.Row
|
||
try:
|
||
yield conn
|
||
finally:
|
||
conn.close()
|
||
|
||
def init_db(self):
|
||
"""初始化数据库"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
|
||
# 内容库表
|
||
cursor.execute('''
|
||
CREATE TABLE IF NOT EXISTS articles (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
product_names TEXT NOT NULL,
|
||
search_title TEXT,
|
||
category TEXT,
|
||
keywords TEXT,
|
||
summary TEXT,
|
||
content TEXT,
|
||
source TEXT,
|
||
url TEXT,
|
||
fetch_date DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
|
||
)
|
||
''')
|
||
|
||
# 为旧表添加字段（如果不存在）
|
||
try:
|
||
cursor.execute('ALTER TABLE articles ADD COLUMN search_title TEXT')
|
||
except:
|
||
pass
|
||
|
||
# 待处理产品列表
|
||
cursor.execute('''
|
||
CREATE TABLE IF NOT EXISTS pending_products (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
product_name TEXT NOT NULL UNIQUE,
|
||
category TEXT,
|
||
subcategory TEXT,
|
||
priority INTEGER DEFAULT 0,
|
||
source TEXT,
|
||
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
|
||
)
|
||
''')
|
||
|
||
# 处理中产品列表
|
||
cursor.execute('''
|
||
CREATE TABLE IF NOT EXISTS processing_products (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
product_name TEXT NOT NULL UNIQUE,
|
||
category TEXT,
|
||
subcategory TEXT,
|
||
status TEXT DEFAULT 'processing',
|
||
started_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||
error_message TEXT
|
||
)
|
||
''')
|
||
|
||
# 处理历史记录
|
||
cursor.execute('''
|
||
CREATE TABLE IF NOT EXISTS process_history (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
product_name TEXT NOT NULL,
|
||
category TEXT,
|
||
subcategory TEXT,
|
||
status TEXT,
|
||
review_id TEXT,
|
||
submitted_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||
details TEXT
|
||
)
|
||
''')
|
||
|
||
# 任务配置表
|
||
cursor.execute('''
|
||
CREATE TABLE IF NOT EXISTS task_configs (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
name TEXT NOT NULL UNIQUE,
|
||
config TEXT,
|
||
enabled INTEGER DEFAULT 1,
|
||
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
|
||
)
|
||
''')
|
||
|
||
# 系统配置表
|
||
cursor.execute('''
|
||
CREATE TABLE IF NOT EXISTS system_config (
|
||
key TEXT PRIMARY KEY,
|
||
value TEXT,
|
||
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
|
||
)
|
||
''')
|
||
|
||
# 失败的URL记录表
|
||
cursor.execute('''
|
||
CREATE TABLE IF NOT EXISTS failed_urls (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
url TEXT NOT NULL,
|
||
title TEXT,
|
||
error_message TEXT,
|
||
retry_count INTEGER DEFAULT 0,
|
||
status TEXT DEFAULT 'failed',
|
||
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||
last_retry_at DATETIME,
|
||
source TEXT DEFAULT 'search'
|
||
)
|
||
''')
|
||
|
||
# 搜索缓存表
|
||
cursor.execute('''
|
||
CREATE TABLE IF NOT EXISTS search_cache (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
keyword TEXT NOT NULL,
|
||
engine TEXT DEFAULT 'bing_cn',
|
||
results TEXT NOT NULL,
|
||
result_count INTEGER,
|
||
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||
expires_at DATETIME,
|
||
UNIQUE(keyword, engine)
|
||
)
|
||
''')
|
||
|
||
# 后台任务表
|
||
cursor.execute('''
|
||
CREATE TABLE IF NOT EXISTS background_tasks (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
task_id TEXT NOT NULL UNIQUE,
|
||
task_type TEXT NOT NULL,
|
||
status TEXT DEFAULT 'pending',
|
||
params TEXT,
|
||
progress INTEGER DEFAULT 0,
|
||
total INTEGER DEFAULT 0,
|
||
current_item TEXT,
|
||
result TEXT,
|
||
error_message TEXT,
|
||
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||
started_at DATETIME,
|
||
finished_at DATETIME
|
||
)
|
||
''')
|
||
|
||
# 处理步骤表
|
||
cursor.execute('''
|
||
CREATE TABLE IF NOT EXISTS process_steps (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
process_id TEXT NOT NULL,
|
||
product_name TEXT NOT NULL,
|
||
step_number INTEGER NOT NULL,
|
||
step_name TEXT NOT NULL,
|
||
step_status TEXT DEFAULT 'pending',
|
||
step_data TEXT,
|
||
started_at DATETIME,
|
||
finished_at DATETIME,
|
||
duration_ms INTEGER,
|
||
error_message TEXT,
|
||
requires_intervention INTEGER DEFAULT 0,
|
||
intervention_type TEXT,
|
||
intervention_status TEXT,
|
||
intervention_data TEXT,
|
||
created_at DATETIME DEFAULT CURRENT_TIMESTAMP
|
||
)
|
||
''')
|
||
|
||
# 处理会话表（用于跟踪整个处理过程）
|
||
cursor.execute('''
|
||
CREATE TABLE IF NOT EXISTS process_sessions (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
session_id TEXT NOT NULL UNIQUE,
|
||
product_name TEXT NOT NULL,
|
||
category TEXT,
|
||
subcategory TEXT,
|
||
status TEXT DEFAULT 'pending',
|
||
current_step INTEGER DEFAULT 0,
|
||
total_steps INTEGER DEFAULT 6,
|
||
paused INTEGER DEFAULT 0,
|
||
pause_reason TEXT,
|
||
result TEXT,
|
||
review_id TEXT,
|
||
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||
started_at DATETIME,
|
||
finished_at DATETIME
|
||
)
|
||
''')
|
||
|
||
# 创建索引
|
||
cursor.execute('CREATE INDEX IF NOT EXISTS idx_process_steps_session ON process_steps(process_id)')
|
||
cursor.execute('CREATE INDEX IF NOT EXISTS idx_process_sessions_status ON process_sessions(status)')
|
||
|
||
conn.commit()
|
||
|
||
# ========== 内容库操作 ==========
|
||
def add_article(self, product_names, category, keywords, summary, content, source, url=None, search_title=None):
|
||
"""添加文章到内容库"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
INSERT INTO articles (product_names, search_title, category, keywords, summary, content, source, url)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
||
''', (json.dumps(product_names, ensure_ascii=False), search_title,
|
||
category, json.dumps(keywords, ensure_ascii=False), summary, content, source, url))
|
||
conn.commit()
|
||
return cursor.lastrowid
|
||
|
||
def search_articles(self, keyword, category=None):
|
||
"""搜索文章"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
if category:
|
||
cursor.execute('''
|
||
SELECT * FROM articles
|
||
WHERE (product_names LIKE ? OR keywords LIKE ? OR summary LIKE ? OR content LIKE ? OR url LIKE ?)
|
||
AND category = ?
|
||
ORDER BY fetch_date DESC
|
||
''', (f'%{keyword}%', f'%{keyword}%', f'%{keyword}%', f'%{keyword}%', f'%{keyword}%', category))
|
||
else:
|
||
cursor.execute('''
|
||
SELECT * FROM articles
|
||
WHERE product_names LIKE ? OR keywords LIKE ? OR summary LIKE ? OR content LIKE ? OR url LIKE ?
|
||
ORDER BY fetch_date DESC
|
||
''', (f'%{keyword}%', f'%{keyword}%', f'%{keyword}%', f'%{keyword}%', f'%{keyword}%'))
|
||
return [dict(row) for row in cursor.fetchall()]
|
||
|
||
def get_article_by_id(self, article_id):
|
||
"""获取文章详情"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('SELECT * FROM articles WHERE id = ?', (article_id,))
|
||
row = cursor.fetchone()
|
||
return dict(row) if row else None
|
||
|
||
def get_all_articles(self, limit=100, offset=0):
|
||
"""获取所有文章"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('SELECT * FROM articles ORDER BY fetch_date DESC LIMIT ? OFFSET ?', (limit, offset))
|
||
return [dict(row) for row in cursor.fetchall()]
|
||
|
||
def delete_article(self, article_id):
|
||
"""删除文章"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('DELETE FROM articles WHERE id = ?', (article_id,))
|
||
conn.commit()
|
||
return cursor.rowcount > 0
|
||
|
||
def get_articles_count(self):
|
||
"""获取文章总数"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('SELECT COUNT(*) as count FROM articles')
|
||
row = cursor.fetchone()
|
||
return row['count'] if row else 0
|
||
|
||
# ========== 待处理产品操作 ==========
|
||
def add_pending_product(self, product_name, category=None, subcategory=None, priority=0, source='manual'):
|
||
"""添加待处理产品"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
try:
|
||
cursor.execute('''
|
||
INSERT INTO pending_products (product_name, category, subcategory, priority, source)
|
||
VALUES (?, ?, ?, ?, ?)
|
||
''', (product_name, category, subcategory, priority, source))
|
||
conn.commit()
|
||
return cursor.lastrowid
|
||
except sqlite3.IntegrityError:
|
||
# 产品已存在，更新优先级
|
||
cursor.execute('''
|
||
UPDATE pending_products
|
||
SET priority = MAX(priority, ?), updated_at = CURRENT_TIMESTAMP
|
||
WHERE product_name = ?
|
||
''', (priority, product_name))
|
||
conn.commit()
|
||
return None
|
||
|
||
def get_pending_products(self, limit=10, order_by='priority'):
|
||
"""获取待处理产品列表"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
if order_by == 'priority':
|
||
cursor.execute('SELECT * FROM pending_products ORDER BY priority DESC, created_at ASC LIMIT ?', (limit,))
|
||
else:
|
||
cursor.execute('SELECT * FROM pending_products ORDER BY created_at ASC LIMIT ?', (limit,))
|
||
return [dict(row) for row in cursor.fetchall()]
|
||
|
||
def get_pending_count(self):
|
||
"""获取待处理产品数量"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('SELECT COUNT(*) FROM pending_products')
|
||
return cursor.fetchone()[0]
|
||
|
||
def remove_pending_product(self, product_name):
|
||
"""从待处理列表移除产品"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('DELETE FROM pending_products WHERE product_name = ?', (product_name,))
|
||
conn.commit()
|
||
return cursor.rowcount > 0
|
||
|
||
# ========== 处理中产品操作 ==========
|
||
def start_processing(self, product_name, category, subcategory):
|
||
"""开始处理产品"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
try:
|
||
cursor.execute('''
|
||
INSERT INTO processing_products (product_name, category, subcategory, status)
|
||
VALUES (?, ?, ?, 'processing')
|
||
''', (product_name, category, subcategory))
|
||
conn.commit()
|
||
return cursor.lastrowid
|
||
except sqlite3.IntegrityError:
|
||
return None
|
||
|
||
def finish_processing(self, product_name, status='completed', error_message=None):
|
||
"""完成处理"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('DELETE FROM processing_products WHERE product_name = ?', (product_name,))
|
||
conn.commit()
|
||
return cursor.rowcount > 0
|
||
|
||
def get_processing_products(self):
|
||
"""获取处理中的产品"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('SELECT * FROM processing_products')
|
||
return [dict(row) for row in cursor.fetchall()]
|
||
|
||
# ========== 处理历史操作 ==========
|
||
def add_process_history(self, product_name, category, subcategory, status, review_id=None, details=None):
|
||
"""添加处理历史"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
INSERT INTO process_history (product_name, category, subcategory, status, review_id, details)
|
||
VALUES (?, ?, ?, ?, ?, ?)
|
||
''', (product_name, category, subcategory, status, review_id,
|
||
json.dumps(details, ensure_ascii=False) if details else None))
|
||
conn.commit()
|
||
return cursor.lastrowid
|
||
|
||
def get_process_history(self, limit=100):
|
||
"""获取处理历史"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('SELECT * FROM process_history ORDER BY submitted_at DESC LIMIT ?', (limit,))
|
||
return [dict(row) for row in cursor.fetchall()]
|
||
|
||
def get_history_by_product(self, product_name):
|
||
"""获取指定产品的处理历史"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('SELECT * FROM process_history WHERE product_name = ? ORDER BY submitted_at DESC', (product_name,))
|
||
return [dict(row) for row in cursor.fetchall()]
|
||
|
||
# ========== 任务配置操作 ==========
|
||
def save_task_config(self, name, config):
|
||
"""保存任务配置"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
INSERT OR REPLACE INTO task_configs (name, config, updated_at)
|
||
VALUES (?, ?, CURRENT_TIMESTAMP)
|
||
''', (name, json.dumps(config, ensure_ascii=False)))
|
||
conn.commit()
|
||
|
||
def get_task_config(self, name):
|
||
"""获取任务配置"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('SELECT * FROM task_configs WHERE name = ?', (name,))
|
||
row = cursor.fetchone()
|
||
if row:
|
||
result = dict(row)
|
||
result['config'] = json.loads(result['config'])
|
||
return result
|
||
return None
|
||
|
||
# ========== 系统配置操作 ==========
|
||
def get_system_config(self, key, default=None):
|
||
"""获取系统配置"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('SELECT value FROM system_config WHERE key = ?', (key,))
|
||
row = cursor.fetchone()
|
||
return row['value'] if row else default
|
||
|
||
def set_system_config(self, key, value):
|
||
"""设置系统配置"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
INSERT OR REPLACE INTO system_config (key, value, updated_at)
|
||
VALUES (?, ?, CURRENT_TIMESTAMP)
|
||
''', (key, value))
|
||
conn.commit()
|
||
|
||
# ========== 失败URL操作 ==========
|
||
def add_failed_url(self, url, title=None, error_message=None, source='search'):
|
||
"""添加失败的URL"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
# 先检查是否已存在
|
||
cursor.execute('SELECT id, retry_count FROM failed_urls WHERE url = ?', (url,))
|
||
existing = cursor.fetchone()
|
||
|
||
if existing:
|
||
# 更新重试次数和错误信息
|
||
cursor.execute('''
|
||
UPDATE failed_urls
|
||
SET error_message = ?, last_retry_at = CURRENT_TIMESTAMP, retry_count = retry_count + 1
|
||
WHERE url = ?
|
||
''', (error_message, url))
|
||
else:
|
||
# 新增失败记录
|
||
cursor.execute('''
|
||
INSERT INTO failed_urls (url, title, error_message, source)
|
||
VALUES (?, ?, ?, ?)
|
||
''', (url, title, error_message, source))
|
||
conn.commit()
|
||
return cursor.lastrowid
|
||
|
||
def get_failed_urls(self, limit=100, status='failed'):
|
||
"""获取失败的URL列表"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
SELECT * FROM failed_urls
|
||
WHERE status = ?
|
||
ORDER BY created_at DESC
|
||
LIMIT ?
|
||
''', (status, limit))
|
||
return [dict(row) for row in cursor.fetchall()]
|
||
|
||
def get_failed_url_count(self):
|
||
"""获取失败URL数量"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('SELECT COUNT(*) FROM failed_urls WHERE status = "failed"')
|
||
return cursor.fetchone()[0]
|
||
|
||
def mark_url_success(self, url):
|
||
"""标记URL为成功（已处理）"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
UPDATE failed_urls
|
||
SET status = 'success', last_retry_at = CURRENT_TIMESTAMP
|
||
WHERE url = ?
|
||
''', (url,))
|
||
conn.commit()
|
||
return cursor.rowcount > 0
|
||
|
||
def delete_failed_url(self, url_id):
|
||
"""删除失败URL记录"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('DELETE FROM failed_urls WHERE id = ?', (url_id,))
|
||
conn.commit()
|
||
return cursor.rowcount > 0
|
||
|
||
def clear_failed_urls(self):
|
||
"""清空所有失败URL记录"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('DELETE FROM failed_urls WHERE status = "failed"')
|
||
conn.commit()
|
||
return cursor.rowcount
|
||
|
||
# ========== 搜索缓存操作 ==========
|
||
def save_search_cache(self, keyword, engine, results, expire_days=7):
|
||
"""保存搜索结果缓存"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
INSERT OR REPLACE INTO search_cache (keyword, engine, results, result_count, created_at, expires_at)
|
||
VALUES (?, ?, ?, ?, CURRENT_TIMESTAMP, datetime('now', '+' || ? || ' days'))
|
||
''', (keyword, engine, json.dumps(results, ensure_ascii=False), len(results), expire_days))
|
||
conn.commit()
|
||
return cursor.lastrowid
|
||
|
||
def get_search_cache(self, keyword, engine='bing_cn'):
|
||
"""获取搜索结果缓存"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
SELECT results, result_count, created_at, expires_at
|
||
FROM search_cache
|
||
WHERE keyword = ? AND engine = ? AND expires_at > datetime('now')
|
||
''', (keyword, engine))
|
||
row = cursor.fetchone()
|
||
if row:
|
||
return {
|
||
'results': json.loads(row['results']),
|
||
'count': row['result_count'],
|
||
'cached_at': row['created_at'],
|
||
'expires_at': row['expires_at']
|
||
}
|
||
return None
|
||
|
||
def clear_expired_cache(self):
|
||
"""清理过期缓存"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('DELETE FROM search_cache WHERE expires_at <= datetime("now")')
|
||
conn.commit()
|
||
return cursor.rowcount
|
||
|
||
def clear_all_cache(self):
|
||
"""清空所有缓存"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('DELETE FROM search_cache')
|
||
conn.commit()
|
||
return cursor.rowcount
|
||
|
||
# ========== 后台任务操作 ==========
|
||
def create_task(self, task_id, task_type, params=None):
|
||
"""创建后台任务"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
INSERT INTO background_tasks (task_id, task_type, params, status)
|
||
VALUES (?, ?, ?, 'pending')
|
||
''', (task_id, task_type, json.dumps(params, ensure_ascii=False) if params else None))
|
||
conn.commit()
|
||
return task_id
|
||
|
||
def get_task(self, task_id):
|
||
"""获取任务详情"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('SELECT * FROM background_tasks WHERE task_id = ?', (task_id,))
|
||
row = cursor.fetchone()
|
||
if row:
|
||
result = dict(row)
|
||
if result.get('params'):
|
||
result['params'] = json.loads(result['params'])
|
||
if result.get('result'):
|
||
result['result'] = json.loads(result['result'])
|
||
return result
|
||
return None
|
||
|
||
def get_active_tasks(self, task_type=None):
|
||
"""获取活动任务（running 或 pending）"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
if task_type:
|
||
cursor.execute('''
|
||
SELECT * FROM background_tasks
|
||
WHERE status IN ('pending', 'running') AND task_type = ?
|
||
ORDER BY created_at DESC
|
||
''', (task_type,))
|
||
else:
|
||
cursor.execute('''
|
||
SELECT * FROM background_tasks
|
||
WHERE status IN ('pending', 'running')
|
||
ORDER BY created_at DESC
|
||
''')
|
||
return [dict(row) for row in cursor.fetchall()]
|
||
|
||
def update_task_status(self, task_id, status, **kwargs):
|
||
"""更新任务状态"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
|
||
updates = ['status = ?']
|
||
values = [status]
|
||
|
||
if status == 'running' and 'started_at' not in kwargs:
|
||
updates.append('started_at = CURRENT_TIMESTAMP')
|
||
elif status in ('completed', 'failed', 'stopped'):
|
||
updates.append('finished_at = CURRENT_TIMESTAMP')
|
||
|
||
for key in ['progress', 'total', 'current_item', 'error_message']:
|
||
if key in kwargs:
|
||
updates.append(f'{key} = ?')
|
||
values.append(kwargs[key])
|
||
|
||
if 'result' in kwargs:
|
||
updates.append('result = ?')
|
||
values.append(json.dumps(kwargs['result'], ensure_ascii=False))
|
||
|
||
values.append(task_id)
|
||
|
||
cursor.execute(
|
||
f'UPDATE background_tasks SET {" , ".join(updates)} WHERE task_id = ?',
|
||
values
|
||
)
|
||
conn.commit()
|
||
return cursor.rowcount > 0
|
||
|
||
def stop_task(self, task_id):
|
||
"""停止任务"""
|
||
return self.update_task_status(task_id, 'stopped')
|
||
|
||
def get_recent_tasks(self, limit=20):
|
||
"""获取最近的任务"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
SELECT * FROM background_tasks
|
||
ORDER BY created_at DESC
|
||
LIMIT ?
|
||
''', (limit,))
|
||
return [dict(row) for row in cursor.fetchall()]
|
||
|
||
def delete_task(self, task_id):
|
||
"""删除任务记录"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('DELETE FROM background_tasks WHERE task_id = ?', (task_id,))
|
||
conn.commit()
|
||
return cursor.rowcount > 0
|
||
|
||
def clear_completed_tasks(self):
|
||
"""清理已完成的任务"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('DELETE FROM background_tasks WHERE status IN ("completed", "failed", "stopped")')
|
||
conn.commit()
|
||
return cursor.rowcount
|
||
|
||
# ========== 处理会话操作 ==========
|
||
def create_process_session(self, session_id, product_name, category=None, subcategory=None):
|
||
"""创建处理会话"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
INSERT INTO process_sessions (session_id, product_name, category, subcategory, status)
|
||
VALUES (?, ?, ?, ?, 'pending')
|
||
''', (session_id, product_name, category, subcategory))
|
||
conn.commit()
|
||
return session_id
|
||
|
||
def get_process_session(self, session_id):
|
||
"""获取处理会话"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('SELECT * FROM process_sessions WHERE session_id = ?', (session_id,))
|
||
row = cursor.fetchone()
|
||
return dict(row) if row else None
|
||
|
||
def update_session_status(self, session_id, status, **kwargs):
|
||
"""更新会话状态"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
|
||
updates = ['status = ?']
|
||
values = [status]
|
||
|
||
if status == 'running' and 'started_at' not in kwargs:
|
||
updates.append('started_at = CURRENT_TIMESTAMP')
|
||
elif status in ('completed', 'failed', 'stopped'):
|
||
updates.append('finished_at = CURRENT_TIMESTAMP')
|
||
|
||
for key in ['current_step', 'result', 'review_id', 'paused', 'pause_reason']:
|
||
if key in kwargs:
|
||
updates.append(f'{key} = ?')
|
||
values.append(kwargs[key])
|
||
|
||
values.append(session_id)
|
||
|
||
cursor.execute(
|
||
f'UPDATE process_sessions SET {" , ".join(updates)} WHERE session_id = ?',
|
||
values
|
||
)
|
||
conn.commit()
|
||
return cursor.rowcount > 0
|
||
|
||
def pause_session(self, session_id, reason=None):
|
||
"""暂停会话"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
UPDATE process_sessions
|
||
SET paused = 1, pause_reason = ?, status = 'paused'
|
||
WHERE session_id = ?
|
||
''', (reason, session_id))
|
||
conn.commit()
|
||
return cursor.rowcount > 0
|
||
|
||
def resume_session(self, session_id):
|
||
"""继续会话"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
UPDATE process_sessions
|
||
SET paused = 0, pause_reason = NULL, status = 'running'
|
||
WHERE session_id = ?
|
||
''', (session_id,))
|
||
conn.commit()
|
||
return cursor.rowcount > 0
|
||
|
||
def get_active_sessions(self):
|
||
"""获取活动的会话"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
SELECT * FROM process_sessions
|
||
WHERE status IN ('pending', 'running', 'paused')
|
||
ORDER BY created_at DESC
|
||
''')
|
||
return [dict(row) for row in cursor.fetchall()]
|
||
|
||
def get_recent_sessions(self, limit=20):
|
||
"""获取最近的会话"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
SELECT * FROM process_sessions
|
||
ORDER BY created_at DESC
|
||
LIMIT ?
|
||
''', (limit,))
|
||
return [dict(row) for row in cursor.fetchall()]
|
||
|
||
# ========== 处理步骤操作 ==========
|
||
def add_process_step(self, process_id, product_name, step_number, step_name, step_data=None):
|
||
"""添加处理步骤"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
INSERT INTO process_steps
|
||
(process_id, product_name, step_number, step_name, step_status, step_data, started_at)
|
||
VALUES (?, ?, ?, ?, 'running', ?, CURRENT_TIMESTAMP)
|
||
''', (process_id, product_name, step_number, step_name,
|
||
json.dumps(step_data, ensure_ascii=False) if step_data else None))
|
||
conn.commit()
|
||
return cursor.lastrowid
|
||
|
||
def update_step_status(self, process_id, step_number, status, **kwargs):
|
||
"""更新步骤状态"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
|
||
updates = ['step_status = ?']
|
||
values = [status]
|
||
|
||
if status in ('completed', 'failed', 'skipped'):
|
||
updates.append('finished_at = CURRENT_TIMESTAMP')
|
||
|
||
for key in ['step_data', 'error_message', 'duration_ms', 'requires_intervention',
|
||
'intervention_type', 'intervention_status', 'intervention_data']:
|
||
if key in kwargs:
|
||
if key in ('step_data', 'intervention_data') and kwargs[key]:
|
||
updates.append(f'{key} = ?')
|
||
values.append(json.dumps(kwargs[key], ensure_ascii=False))
|
||
else:
|
||
updates.append(f'{key} = ?')
|
||
values.append(kwargs[key])
|
||
|
||
values.extend([process_id, step_number])
|
||
|
||
cursor.execute(
|
||
f'UPDATE process_steps SET {" , ".join(updates)} WHERE process_id = ? AND step_number = ?',
|
||
values
|
||
)
|
||
conn.commit()
|
||
return cursor.rowcount > 0
|
||
|
||
def get_process_steps(self, process_id):
|
||
"""获取处理步骤列表"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
SELECT * FROM process_steps
|
||
WHERE process_id = ?
|
||
ORDER BY step_number ASC
|
||
''', (process_id,))
|
||
return [dict(row) for row in cursor.fetchall()]
|
||
|
||
def get_step_detail(self, process_id, step_number):
|
||
"""获取步骤详情"""
|
||
with self.get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
SELECT * FROM process_steps
|
||
WHERE process_id = ? AND step_number = ?
|
||
''', (process_id, step_number))
|
||
row = cursor.fetchone()
|
||
if row:
|
||
result = dict(row)
|
||
if result.get('step_data'):
|
||
result['step_data'] = json.loads(result['step_data'])
|
||
if result.get('intervention_data'):
|
||
result['intervention_data'] = json.loads(result['intervention_data'])
|
||
return result
|
||
return None
|
||
|
||
def set_step_intervention(self, process_id, step_number, intervention_type, intervention_data=None):
|
||
"""设置步骤需要干预"""
|
||
return self.update_step_status(
|
||
process_id, step_number, 'paused',
|
||
requires_intervention=1,
|
||
intervention_type=intervention_type,
|
||
intervention_status='pending',
|
||
intervention_data=intervention_data
|
||
)
|
||
|
||
def complete_intervention(self, process_id, step_number, intervention_data=None):
|
||
"""完成干预"""
|
||
return self.update_step_status(
|
||
process_id, step_number, 'completed',
|
||
requires_intervention=0,
|
||
intervention_status='completed',
|
||
intervention_data=intervention_data
|
||
)
|
||
|
||
# 全局数据库实例
|
||
db = Database() |