AdamWen95
94 lines
4.6 KiB
Python
94 lines
4.6 KiB
Python
"""Refresh reviewed sources atomically without resetting saved options or history."""
|
|
import argparse
|
|
import copy
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
from console import catalog, native_plans
|
|
from console.settings import Settings
|
|
from console.store import Store
|
|
|
|
|
|
def refresh(store, backup, apply=False):
|
|
original = {s['id']: s for s in store.scripts()}
|
|
class Draft:
|
|
settings = store.settings
|
|
items = copy.deepcopy(original)
|
|
def scripts(self): return list(self.items.values())
|
|
def save_script(self, item): self.items[item['id']] = item
|
|
draft = Draft()
|
|
scan = catalog.discover(draft)
|
|
for item in draft.items.values():
|
|
profile = native_plans.profile_for(store.settings, item)
|
|
if not profile:
|
|
continue
|
|
if item['id'] not in original:
|
|
item['config'] = native_plans.validate_config(store.settings, item,
|
|
{'native_options': {}, 'timeout_seconds': profile.get('timeout_seconds', 3600)})
|
|
item['enabled'] = profile.get('available', True)
|
|
if not profile.get('available', True):
|
|
item.update(enabled=False, blocked=True, reason=profile.get('unavailable_reason', '原环境待配置'))
|
|
if item['enabled']:
|
|
item.update(status='ready', reason='')
|
|
elif item['status'] != 'missing':
|
|
item['status'] = 'needs_configuration'
|
|
changes = []
|
|
for identity, value in draft.items.items():
|
|
before = original.get(identity)
|
|
if before and {k:v for k,v in before.items() if k != 'updated_at'} == {k:v for k,v in value.items() if k != 'updated_at'}:
|
|
continue
|
|
changes.append((identity, before, value))
|
|
if not apply:
|
|
return {'scan': scan, 'changed': len(changes), 'applied': False}
|
|
with store.connect(write=True) as db:
|
|
if db.execute("SELECT 1 FROM runs WHERE state IN ('queued','running')").fetchone():
|
|
raise ValueError('请先等待排队和运行任务结束')
|
|
records = []
|
|
for identity, before, after in changes:
|
|
row = db.execute('SELECT data FROM scripts WHERE id=?', (identity,)).fetchone()
|
|
if (json.loads(row[0]) if row else None) != before:
|
|
raise ValueError('配置有并发修改，本次刷新未写入')
|
|
records.append({'id': identity, 'before': row[0] if row else None, 'after': json.dumps(after)})
|
|
if not records:
|
|
return {'scan': scan, 'changed': 0, 'applied': True}
|
|
backup.parent.mkdir(parents=True, exist_ok=True)
|
|
with backup.open('x', encoding='utf-8') as handle:
|
|
json.dump({'scripts': records}, handle, ensure_ascii=False, indent=2)
|
|
backup.chmod(0o600)
|
|
for record in records:
|
|
db.execute('INSERT OR REPLACE INTO scripts VALUES(?,?)', (record['id'], record['after']))
|
|
return {'scan': scan, 'changed': len(changes), 'applied': True}
|
|
|
|
|
|
def restore(store, backup):
|
|
records = json.loads(backup.read_text(encoding='utf-8'))['scripts']
|
|
with store.connect(write=True) as db:
|
|
if db.execute("SELECT 1 FROM runs WHERE state IN ('queued','running')").fetchone():
|
|
raise ValueError('请先等待排队和运行任务结束')
|
|
for record in records:
|
|
row = db.execute('SELECT data FROM scripts WHERE id=?', (record['id'],)).fetchone()
|
|
current = row[0] if row else None
|
|
if current not in (record['before'], record['after']):
|
|
raise ValueError('刷新后配置已修改，请人工核对，禁止覆盖')
|
|
if record['before'] is None and db.execute('SELECT 1 FROM runs WHERE script_id=?', (record['id'],)).fetchone():
|
|
raise ValueError('新增项目已有运行历史，不能删除登记，请人工恢复')
|
|
for record in records:
|
|
if record['before'] is None:
|
|
db.execute('DELETE FROM scripts WHERE id=?', (record['id'],))
|
|
else:
|
|
db.execute('UPDATE scripts SET data=? WHERE id=?', (record['before'], record['id']))
|
|
return {'restored': len(records), 'history_preserved': True}
|
|
|
|
|
|
if __name__ == '__main__':
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument('--apply', action='store_true')
|
|
parser.add_argument('--restore', action='store_true')
|
|
parser.add_argument('--backup', type=Path, required=True)
|
|
args = parser.parse_args()
|
|
if args.apply and args.restore: parser.error('选择刷新或恢复之一')
|
|
store = Store(Settings.from_env())
|
|
print(json.dumps(restore(store, args.backup) if args.restore else refresh(store, args.backup, args.apply)))
|