433 lines
25 KiB
Python
433 lines
25 KiB
Python
"""Persistent serial queue. Browser submissions are always initiated by the user."""
|
|
import json
|
|
import copy
|
|
import re
|
|
from contextlib import contextmanager
|
|
from pathlib import Path
|
|
import secrets
|
|
import sqlite3
|
|
import threading
|
|
import time
|
|
from urllib.parse import urlsplit, parse_qs
|
|
|
|
from articles import inspect_article, short_title
|
|
|
|
ACCEPTED = {'published', 'accepted_pending_moderation'}
|
|
SUBMITTED = {'user_submitting', 'submit_unknown'}
|
|
|
|
|
|
class Store:
|
|
def __init__(self, path, ledger=None):
|
|
self.path = str(path)
|
|
Path(path).parent.mkdir(parents=True, exist_ok=True)
|
|
self.lock = threading.RLock()
|
|
with self.db() as db:
|
|
db.executescript('''
|
|
CREATE TABLE IF NOT EXISTS jobs(id TEXT PRIMARY KEY, article_key TEXT UNIQUE, value TEXT NOT NULL);
|
|
CREATE TABLE IF NOT EXISTS ledger(key TEXT PRIMARY KEY, value TEXT NOT NULL);
|
|
CREATE TABLE IF NOT EXISTS settings(id INTEGER PRIMARY KEY CHECK(id=1), value TEXT NOT NULL);
|
|
CREATE TABLE IF NOT EXISTS posts(id TEXT PRIMARY KEY, job TEXT NOT NULL);
|
|
CREATE TABLE IF NOT EXISTS post_aliases(url TEXT PRIMARY KEY, thread_id TEXT NOT NULL, job TEXT NOT NULL);
|
|
''')
|
|
row = db.execute('SELECT value FROM settings WHERE id=1').fetchone()
|
|
self.run = json.loads(row[0]) if row else dict(running=False, epoch=0, config={}, batch=None, active=None, owner=None, switch=None, reason='请配置任务后开始')
|
|
self.run.update(running=False, epoch=self.run['epoch'] + 1, reason='服务已启动，请明确点击开始/继续；未决任务不会重发')
|
|
for row in db.execute('SELECT value FROM jobs').fetchall():
|
|
job = json.loads(row[0])
|
|
if job['state'] in ACCEPTED and job.get('post_url'):
|
|
canonical = job.get('thread_id')
|
|
if not canonical:
|
|
job['legacy_url_id'] = job.get('post_id')
|
|
job['identity_needs_reconciliation'] = True
|
|
self._save(db, job)
|
|
db.execute('INSERT OR IGNORE INTO post_aliases VALUES(?,?,?)',
|
|
(job['post_url'], canonical or 'unknown', job['id']))
|
|
self._persist(db)
|
|
self.clients = {}
|
|
if ledger and Path(ledger).is_file():
|
|
self.import_history(json.loads(Path(ledger).read_text()).get('records', []))
|
|
|
|
@contextmanager
|
|
def db(self):
|
|
connection = sqlite3.connect(self.path, timeout=20)
|
|
previous = copy.deepcopy(getattr(self, 'run', None))
|
|
try:
|
|
with connection:
|
|
yield connection
|
|
except Exception:
|
|
if previous is not None:
|
|
self.run = previous
|
|
raise
|
|
finally:
|
|
connection.close()
|
|
|
|
def _persist(self, db):
|
|
db.execute('INSERT OR REPLACE INTO settings VALUES(1,?)', (json.dumps(self.run, ensure_ascii=False),))
|
|
|
|
def _job(self, db, id):
|
|
row = db.execute('SELECT value FROM jobs WHERE id=?', (id,)).fetchone()
|
|
if not row:
|
|
raise ValueError('任务不存在')
|
|
return json.loads(row[0])
|
|
|
|
def _save(self, db, job):
|
|
job['updated_at'] = time.time()
|
|
db.execute('INSERT OR REPLACE INTO jobs VALUES(?,?,?)', (job['id'], job['article_key'], json.dumps(job, ensure_ascii=False)))
|
|
|
|
def import_history(self, records):
|
|
with self.lock, self.db() as db:
|
|
for record in records:
|
|
digest = record.get('markdown_sha256')
|
|
if digest and record.get('block_repost'):
|
|
db.execute('INSERT OR IGNORE INTO ledger VALUES(?,?)', ('md:' + digest, json.dumps(record, ensure_ascii=False)))
|
|
# Add content identity where the original historical file is still present.
|
|
try:
|
|
article = inspect_article(Path(record['markdown_path']).parent)
|
|
db.execute('INSERT OR IGNORE INTO ledger VALUES(?,?)', ('article:' + article['article_key'], json.dumps(record, ensure_ascii=False)))
|
|
except (OSError, ValueError, KeyError):
|
|
pass
|
|
|
|
def enqueue(self, directory):
|
|
article = inspect_article(directory)
|
|
article.pop('package')
|
|
keys = ['article:' + article['article_key'], 'md:' + article['markdown_sha256']] + ['source:' + str(k) for k in article['source_keys']]
|
|
with self.lock, self.db() as db:
|
|
if any(db.execute('SELECT 1 FROM ledger WHERE key=?', (k,)).fetchone() for k in keys):
|
|
return '已受理，跳过'
|
|
if db.execute('SELECT 1 FROM jobs WHERE article_key=?', (article['article_key'],)).fetchone():
|
|
return '已在队列，跳过'
|
|
# Source URL identities prevent different renderings of one downloaded article re-entering.
|
|
for row in db.execute('SELECT value FROM jobs'):
|
|
if set(article['source_keys']) & set(json.loads(row[0]).get('source_keys', [])):
|
|
return '相同来源已在队列，跳过'
|
|
article.update(id=secrets.token_hex(12), state='queued', created_at=time.time(), error='', post_url='')
|
|
self._save(db, article)
|
|
return '已入队'
|
|
|
|
def state(self):
|
|
with self.lock, self.db() as db:
|
|
jobs = [json.loads(r[0]) for r in db.execute('SELECT value FROM jobs ORDER BY rowid')]
|
|
return dict(run=json.loads(json.dumps(self.run)), jobs=jobs,
|
|
clients=[v for v in self.clients.values() if time.time() - v['seen'] < 20],
|
|
history_count=db.execute('SELECT count(*) FROM ledger').fetchone()[0])
|
|
|
|
def configure(self, config):
|
|
fid, minimum, maximum = config.get('fid'), config.get('minimum'), config.get('maximum')
|
|
interval = config.get('interval')
|
|
if any(type(x) is not int for x in (fid, minimum, maximum, interval)) or not (fid > 0 and 1 <= minimum <= maximum <= 1000 and 0 <= interval <= 86400):
|
|
raise ValueError('请填写有效版块 ID、篇数范围（1～1000）和间隔秒数')
|
|
if config.get('history_checked') is not True:
|
|
raise ValueError('请先核对待处理文章与历史发布记录，并勾选确认')
|
|
with self.lock, self.db() as db:
|
|
if self.run['active'] or self.run['switch'] or self.run['running']:
|
|
raise ValueError('当前流程未结束，不能更换配置')
|
|
self.run.update(config={k: config[k] for k in ('fid', 'minimum', 'maximum', 'interval', 'history_checked')}, batch=None)
|
|
self._persist(db)
|
|
|
|
def control(self, action):
|
|
with self.lock, self.db() as db:
|
|
if action == 'start':
|
|
if not self.run['config']:
|
|
raise ValueError('请先保存配置')
|
|
self.run.update(running=True, reason='等待浏览器连接')
|
|
else:
|
|
self.run.update(running=False, reason='已终止' if action == 'stop' else '已暂停')
|
|
self.run['epoch'] += 1
|
|
self._persist(db)
|
|
return dict(self.run)
|
|
|
|
def _batch(self, account):
|
|
cfg = self.run['config']
|
|
return dict(account=account, target=cfg['minimum'] + secrets.randbelow(cfg['maximum'] - cfg['minimum'] + 1), count=0)
|
|
|
|
def _check_owner(self, owner):
|
|
if self.run['owner'] != owner:
|
|
raise ValueError('任务由另一个浏览器标签持有')
|
|
|
|
def heartbeat(self, data):
|
|
owner = str(data.get('owner', ''))[:100]
|
|
if not owner:
|
|
raise ValueError('缺少执行标签标识')
|
|
with self.lock:
|
|
self.clients[owner] = dict(owner=owner, account=data.get('account'), url=str(data.get('url', ''))[:1000],
|
|
extension=data.get('extension'), seen=time.time())
|
|
return self.state()['run']
|
|
|
|
def next(self, data):
|
|
owner, account = data.get('owner'), data.get('account')
|
|
with self.lock, self.db() as db:
|
|
if not self.run['running']:
|
|
return dict(action='idle', run=self.run)
|
|
if not account or not str(account.get('uid', '')).isdigit() or not account.get('name'):
|
|
return dict(action='login', run=self.run)
|
|
if self.run['owner'] not in (None, owner):
|
|
return dict(action='other_tab', run=self.run)
|
|
self.run['owner'] = owner
|
|
if self.run['switch']:
|
|
switch = self.run['switch']
|
|
ext = data.get('extension') or {}
|
|
if ext.get('ready') is not True:
|
|
self.run.update(running=False, reason='换号插件联动接口未就绪，请在原安装目录更新并重新加载插件、刷新网站后继续；未重新抽号')
|
|
self._persist(db)
|
|
return dict(action='idle', run=self.run)
|
|
if account['uid'] != switch['from'] and ext.get('idle') is True:
|
|
self.run.update(batch=self._batch(account), switch=None)
|
|
self._persist(db)
|
|
else:
|
|
# Pending means no click was issued. Time spent stopped, or
|
|
# waiting for the plugin, is not a failed login attempt.
|
|
elapsed = time.time() - switch.get('clicked_at', switch['created_at'])
|
|
if ext.get('paused') or (switch['state'] == 'clicked' and (elapsed > 120 or (elapsed > 10 and ext.get('idle')))):
|
|
self.run.update(running=False, reason='换号已暂停、超时或账号未改变，请检查插件并人工恢复，禁止重复抽号')
|
|
self._persist(db)
|
|
return dict(action='idle', run=self.run)
|
|
return dict(action='switch' if switch['state'] == 'pending' else 'switch_wait', run=self.run)
|
|
if self.run['batch'] is None:
|
|
self.run['batch'] = self._batch(account)
|
|
if self.run['batch']['account']['uid'] != account['uid']:
|
|
self.run.update(running=False, reason='检测到账号变化，请先核对当前任务')
|
|
self._persist(db)
|
|
return dict(action='idle', run=self.run)
|
|
if self.run.get('return_home'):
|
|
url = urlsplit(str(data.get('url', '')))
|
|
if url.scheme == 'https' and url.netloc in ('www.dww11.com', 'dww11.com') and url.path == '/' and not url.query:
|
|
self.run['return_home'] = False
|
|
self._persist(db)
|
|
else:
|
|
return dict(action='home', run=self.run)
|
|
if self.run['active']:
|
|
job = self._job(db, self.run['active'])
|
|
return dict(action='resume', job=job, run=self.run)
|
|
rows = [json.loads(r[0]) for r in db.execute('SELECT value FROM jobs ORDER BY rowid')]
|
|
job = next((j for j in rows if j['state'] == 'queued'), None)
|
|
if not job:
|
|
self.run.update(running=False, reason='当前队列已处理完；新文章入队后请点击继续')
|
|
self._persist(db)
|
|
return dict(action='idle', run=self.run)
|
|
if time.time() < self.run.get('not_before', 0):
|
|
return dict(action='interval', run=self.run)
|
|
if self.run['batch']['count'] >= self.run['batch']['target']:
|
|
self.run['switch'] = dict(state='pending', **{'from': account['uid']}, created_at=time.time())
|
|
self._persist(db)
|
|
return dict(action='switch', run=self.run)
|
|
job['title'] = short_title(job.get('original_title', job['title']))
|
|
job.update(state='preparing', account=account, fid=self.run['config']['fid'], error='')
|
|
self.run['active'] = job['id']
|
|
self._save(db, job)
|
|
self._persist(db)
|
|
return dict(action='prepare', job=job, run=self.run)
|
|
|
|
def current(self, data):
|
|
with self.lock, self.db() as db:
|
|
self._check_owner(data['owner'])
|
|
return {'job': self._job(db, self.run['active']) if self.run['active'] else None, 'run': self.run}
|
|
|
|
def skip(self, data):
|
|
"""Explicitly discard an unsubmitted task and continue the existing queue."""
|
|
with self.lock, self.db() as db:
|
|
self._check_owner(data['owner'])
|
|
if data.get('epoch') != self.run['epoch']:
|
|
raise ValueError('流程状态已改变，请重新确认后跳过')
|
|
if self.run['active'] != data.get('id'):
|
|
raise ValueError('不是当前任务，不能跳过')
|
|
job = self._job(db, data['id'])
|
|
if job['state'] not in {'preparing', 'importing', 'formatting', 'awaiting_user_review', 'needs_attention'} or job.get('submitted_at'):
|
|
raise ValueError('文章已提交或结果待核对，不能直接跳过，请先人工核对')
|
|
job.update(state='skipped', error='', skipped_at=time.time())
|
|
self.run.update(active=None, running=True, epoch=self.run['epoch'] + 1,
|
|
reason='已跳过当前文章，等待准备下一篇')
|
|
self._save(db, job)
|
|
self._persist(db)
|
|
return {'ok': True, 'fid': job['fid'], 'run': dict(self.run)}
|
|
|
|
def switch_begin(self, data):
|
|
with self.lock, self.db() as db:
|
|
self._check_owner(data['owner'])
|
|
if not self.run['running'] or not self.run['switch'] or self.run['switch']['state'] != 'pending':
|
|
raise ValueError('换号状态已改变，禁止重复点击')
|
|
if data.get('epoch', self.run['epoch']) != self.run['epoch']:
|
|
raise ValueError('换号请求已过期')
|
|
self.run['switch'].update(state='clicked', clicked_at=time.time())
|
|
self._persist(db)
|
|
return {'ok': True}
|
|
|
|
def package(self, data):
|
|
with self.lock, self.db() as db:
|
|
self._check_owner(data['owner'])
|
|
if not self.run['running'] or self.run['active'] != data['id']:
|
|
raise ValueError('流程已停止或任务不匹配')
|
|
job = self._job(db, data['id'])
|
|
if job['state'] != 'preparing':
|
|
raise ValueError('当前任务不允许重新导入，请先核对已上传内容')
|
|
article = inspect_article(job['directory'])
|
|
if article['markdown_sha256'] != job['markdown_sha256'] or article['article_key'] != job['article_key']:
|
|
raise ValueError('入队后原文章已改变，请移出并重新扫描')
|
|
return article['package']
|
|
|
|
def event(self, data):
|
|
with self.lock, self.db() as db:
|
|
self._check_owner(data['owner'])
|
|
job = self._job(db, data['id'])
|
|
if self.run['active'] != job['id']:
|
|
raise ValueError('不是当前任务')
|
|
target = data['state']
|
|
transitions = {'preparing': {'importing', 'needs_attention'}, 'importing': {'formatting', 'needs_attention'},
|
|
'formatting': {'awaiting_user_review', 'needs_attention'},
|
|
'awaiting_user_review': {'user_submitting', 'needs_attention'},
|
|
'user_submitting': {'submit_unknown', 'needs_attention', 'rejected'}, 'submit_unknown': {'needs_attention'}}
|
|
if target not in transitions.get(job['state'], set()):
|
|
raise ValueError('任务状态不允许此操作：' + job['state'] + ' → ' + target)
|
|
if target in ('importing', 'formatting', 'awaiting_user_review') and (not self.run['running'] or data.get('epoch') != self.run['epoch']):
|
|
raise ValueError('流程已暂停/终止，过期回调不能继续')
|
|
job.update(state=target, error=str(data.get('error', ''))[:1000])
|
|
if target == 'user_submitting':
|
|
job.update(submitted_at=time.time(), snapshot=data.get('snapshot', {}), attempt=secrets.token_hex(8))
|
|
if target == 'awaiting_user_review':
|
|
job['prepared'] = data.get('snapshot', {})
|
|
if target in ('needs_attention', 'rejected'):
|
|
self.run.update(running=False, reason=job['error'] or '当前任务需要处理')
|
|
self._persist(db)
|
|
self._save(db, job)
|
|
return {'ok': True}
|
|
|
|
def reconcile(self, data):
|
|
if data.get('confirmed') is not True:
|
|
raise ValueError('请先确认这是当前文章的已受理结果')
|
|
with self.lock, self.db() as db:
|
|
if self.run['running']:
|
|
raise ValueError('请先暂停流程再补核历史受理记录')
|
|
job = self._job(db, data['id'])
|
|
if data.get('article_key') != job['article_key']:
|
|
raise ValueError('文章身份不匹配')
|
|
return self._accept({**data, 'owner': self.run['owner']}, count_batch=False)
|
|
|
|
def accept(self, data):
|
|
return self._accept(data)
|
|
|
|
def _accept(self, data, count_batch=True):
|
|
with self.lock, self.db() as db:
|
|
self._check_owner(data['owner'])
|
|
job = self._job(db, data['id'])
|
|
if job['state'] in ACCEPTED:
|
|
return {'ok': True, 'duplicate': True}
|
|
if self.run['active'] != job['id'] or job['state'] not in SUBMITTED:
|
|
raise ValueError('只有已观察到人工提交的任务才能登记结果')
|
|
if isinstance(data.get('evidence'), dict) and data['evidence'].get('adapter') == 'moderation_prompt':
|
|
if not count_batch:
|
|
raise ValueError('审核提示只用于当前人工提交任务，不能补核历史结果')
|
|
return self._accept_prompt(db, job, data)
|
|
url = urlsplit(str(data.get('post_url', '')))
|
|
match = re.fullmatch(r'/thread/(\d+)\.html', url.path)
|
|
query = parse_qs(url.query)
|
|
thread_id = str(data.get('thread_id', ''))
|
|
if not re.fullmatch(r'[1-9]\d*', thread_id):
|
|
raise ValueError('缺少详情页真实主题 ID，不能以地址数字代替')
|
|
if url.scheme != 'https' or url.netloc not in ('www.dww11.com', 'dww11.com') or not (match or (url.path == '/forum.php' and query.get('mod') == ['viewthread'] and query.get('tid') == [thread_id])):
|
|
raise ValueError('未识别到帖子详情页')
|
|
if data.get('author_uid') != job['account']['uid'] or data.get('fid') != job['fid']:
|
|
raise ValueError('结果页作者或版块与任务不符，请人工核对')
|
|
if ' '.join(str(data.get('title', '')).split()) != ' '.join(str(job.get('snapshot', {}).get('title', '')).split()):
|
|
raise ValueError('结果页标题与人工提交快照不一致')
|
|
if data.get('state') not in ACCEPTED:
|
|
raise ValueError('需要明确的公开或审核中状态')
|
|
if not data.get('evidence'):
|
|
raise ValueError('缺少页面结果证据')
|
|
evidence = data['evidence']
|
|
if not isinstance(evidence, dict) or str(evidence.get('thread_id')) != thread_id:
|
|
raise ValueError('主题 ID 与页面证据不一致')
|
|
alias = db.execute('SELECT thread_id, job FROM post_aliases WHERE url=?', (url.geturl(),)).fetchone()
|
|
if alias and (alias[0] != thread_id or alias[1] != job['id']):
|
|
raise ValueError('帖子链接已绑定其他记录')
|
|
post_id = thread_id
|
|
existing = db.execute('SELECT job FROM posts WHERE id=?', (post_id,)).fetchone()
|
|
if existing and existing[0] != job['id']:
|
|
raise ValueError('帖子已绑定其他任务，不能重复计数')
|
|
db.execute('INSERT OR IGNORE INTO posts VALUES(?,?)', (post_id, job['id']))
|
|
db.execute('INSERT OR IGNORE INTO post_aliases VALUES(?,?,?)', (url.geturl(), thread_id, job['id']))
|
|
job.update(state=data['state'], thread_id=thread_id, post_id=post_id, first_post_pid=data.get('first_post_pid'), post_url=url.geturl(), result=evidence, integrity_issues=evidence.get('integrity_issues', []))
|
|
keys = ['article:' + job['article_key'], 'md:' + job['markdown_sha256']] + ['source:' + str(k) for k in job['source_keys']]
|
|
for key in keys:
|
|
db.execute('INSERT OR REPLACE INTO ledger VALUES(?,?)', (key, json.dumps(job, ensure_ascii=False)))
|
|
if count_batch:
|
|
self.run['batch']['count'] += 1
|
|
job['historical_reconciliation'] = not count_batch
|
|
self.run.update(active=None, return_home=True, not_before=time.time() + self.run['config']['interval'])
|
|
if job['integrity_issues']:
|
|
self.run.update(running=False, reason='网站已受理，内容核对发现差异，请检查帖子后明确继续')
|
|
self._save(db, job)
|
|
self._persist(db)
|
|
return {'ok': True, 'run': dict(self.run)}
|
|
|
|
def _accept_prompt(self, db, job, data):
|
|
"""Temporary acceptance rule: the forum's explicit moderation prompt only."""
|
|
evidence = data['evidence']
|
|
message = '新主题需要审核，您的帖子通过审核后才能显示'
|
|
text = ' '.join(str(evidence.get('observed_text', '')).split())
|
|
url = urlsplit(str(evidence.get('observed_url', '')))
|
|
if (data.get('state') != 'accepted_pending_moderation'
|
|
or evidence.get('message') != message or message not in text):
|
|
raise ValueError('缺少网站新主题审核提示')
|
|
if url.scheme != 'https' or url.netloc not in ('www.dww11.com', 'dww11.com'):
|
|
raise ValueError('审核提示必须来自论坛网站')
|
|
if not job.get('attempt') or data.get('attempt') != job['attempt']:
|
|
raise ValueError('审核提示与当前提交批次不一致')
|
|
# 未核对详情页，不虚构主题 ID、首帖 ID 或帖子地址。
|
|
job.update(state='accepted_pending_moderation', error='', result=evidence,
|
|
integrity_issues=[], historical_reconciliation=False,
|
|
acceptance_basis='moderation_prompt')
|
|
keys = ['article:' + job['article_key'], 'md:' + job['markdown_sha256']] + ['source:' + str(k) for k in job['source_keys']]
|
|
for key in keys:
|
|
db.execute('INSERT OR REPLACE INTO ledger VALUES(?,?)', (key, json.dumps(job, ensure_ascii=False)))
|
|
self.run['batch']['count'] += 1
|
|
self.run.update(active=None, return_home=True,
|
|
not_before=time.time() + self.run['config']['interval'])
|
|
self._save(db, job)
|
|
self._persist(db)
|
|
return {'ok': True, 'run': dict(self.run)}
|
|
|
|
def resolve(self, data):
|
|
"""Explicit local UI resolution, never an automatic retry of an unknown submission."""
|
|
with self.lock, self.db() as db:
|
|
job = self._job(db, data['id'])
|
|
if job['state'] in ACCEPTED:
|
|
raise ValueError('已受理文章不能重置')
|
|
if self.run['running']:
|
|
raise ValueError('请先暂停流程')
|
|
if data.get('confirmed_not_accepted') is not True:
|
|
raise ValueError('必须先人工核对网站未受理，并清理编辑器/草稿')
|
|
job.update(state='queued' if data.get('retry') else 'cancelled', error='', snapshot={})
|
|
for key in ('submitted_at', 'attempt', 'prepared'):
|
|
job.pop(key, None)
|
|
if self.run['active'] == job['id']:
|
|
self.run.update(active=None, owner=None)
|
|
self._save(db, job)
|
|
self._persist(db)
|
|
return {'ok': True}
|
|
|
|
def remove(self, data):
|
|
"""Remove queue rows only; retain files, acceptance ledger and post bindings."""
|
|
ids = data.get('ids')
|
|
if not isinstance(ids, list) or not ids or not all(isinstance(id, str) and id for id in ids):
|
|
raise ValueError('请选择要移除的文章')
|
|
ids = list(dict.fromkeys(ids))
|
|
with self.lock, self.db() as db:
|
|
if self.run['running']:
|
|
raise ValueError('请先暂停流程，再移除文章')
|
|
# Validate the entire selection before deleting anything.
|
|
for id in ids:
|
|
job = self._job(db, id)
|
|
if id == self.run['active']:
|
|
raise ValueError('当前任务不能直接移除，请先通过“人工核对 / 处理”核对并移出')
|
|
if job['state'] in SUBMITTED or (job.get('submitted_at') and job['state'] not in ACCEPTED):
|
|
raise ValueError('文章提交结果尚未核对，请先核对并处理后再移除')
|
|
db.executemany('DELETE FROM jobs WHERE id=?', [(id,) for id in ids])
|
|
return {'ok': True, 'removed': len(ids)}
|
|
|
|
def takeover(self, data):
|
|
with self.lock, self.db() as db:
|
|
if self.run['running']:
|
|
raise ValueError('请先在控制台暂停流程，再接管')
|
|
self.run['owner'] = data['owner']
|
|
self._persist(db)
|
|
return {'ok': True}
|