"""User-requested timestamp redistribution with an immutable plan and guarded rollback."""
|
from pathlib import Path
|
from datetime import datetime, date, time, timedelta, timezone
|
from collections import Counter
|
import argparse
|
import hashlib
|
import json
|
import random
|
|
from read_log_times import connect
|
|
HERE = Path(__file__).resolve().parent
|
PLAN = HERE / 'time-adjustment-backup-and-plan.json'
|
RESULT = HERE / 'time-adjustment-result.json'
|
TZ = timezone(timedelta(hours=8))
|
|
def digest(rows):
|
# All fields except the user-requested operation time must remain unchanged.
|
return hashlib.sha256(json.dumps([{k: v for k, v in r.items() if k != 'create_time'}
|
for r in rows], sort_keys=True, default=str).encode()).hexdigest()
|
|
def summary(rows):
|
return dict(count=len(rows), earliest=min(r['create_time'] for r in rows),
|
latest=max(r['create_time'] for r in rows),
|
days=dict(sorted(Counter(str(r['create_time'])[:10] for r in rows).items())))
|
|
def save(path, data, exclusive=False):
|
with path.open('x' if exclusive else 'w', encoding='utf-8') as f:
|
json.dump(data, f, ensure_ascii=False, indent=2, default=str)
|
|
def plan():
|
cutoff = datetime.now(TZ).replace(tzinfo=None, microsecond=0)
|
start = date(2026, 9, 17)
|
assert cutoff.date() >= start
|
with connect() as c:
|
with c.cursor() as q:
|
q.execute('SELECT * FROM system_operate_log ORDER BY id')
|
rows = q.fetchall()
|
assert rows, 'No operation logs found'
|
days = [start + timedelta(days=i) for i in range((cutoff.date()-start).days+1)]
|
days = [d for d in days if datetime.combine(d, time(9)) <= cutoff]
|
targets = []
|
rng = random.SystemRandom()
|
for i in range(len(rows)):
|
d = days[i % len(days)]
|
lower = datetime.combine(d, time(9))
|
upper = min(datetime.combine(d, time(18)), cutoff)
|
targets.append(lower + timedelta(seconds=rng.randrange(int((upper-lower).total_seconds())+1)))
|
rng.shuffle(targets)
|
changes = [dict(id=r['id'], old_create_time=str(r['create_time']),
|
old_update_time=str(r['update_time']), new_create_time=str(t))
|
for r, t in zip(rows, targets)]
|
save(PLAN, dict(target='36.213.150.237:10000/mom-hm', table='system_operate_log',
|
purpose='User-requested artificial timestamp distribution; original event times retained here.',
|
timezone='Asia/Shanghai', cutoff=str(cutoff), original_summary=summary(rows),
|
preserved_fields_digest=digest(rows), changes=changes), exclusive=True)
|
print(json.dumps(dict(plan=str(PLAN), count=len(rows), days=len(days), cutoff=str(cutoff)), ensure_ascii=False))
|
|
def apply(restore=False):
|
data = json.loads(PLAN.read_text(encoding='utf-8'))
|
changes = data['changes']
|
ids = [r['id'] for r in changes]
|
marks = ','.join(['%s'] * len(ids))
|
sql = 'SELECT * FROM system_operate_log WHERE id IN ('+marks+') ORDER BY id'
|
with connect() as c:
|
with c.cursor() as q:
|
q.execute(sql+' FOR UPDATE', ids)
|
before = q.fetchall()
|
assert len(before) == len(changes)
|
assert digest(before) == data['preserved_fields_digest'], 'Log fields changed since snapshot'
|
by_id = {r['id']: r for r in before}
|
for change in changes:
|
expected = change['new_create_time'] if restore else change['old_create_time']
|
assert str(by_id[change['id']]['create_time']) == expected, 'Operation time changed since snapshot'
|
params = [(r['old_create_time'] if restore else r['new_create_time'],
|
r['old_update_time'], r['id']) for r in changes]
|
# Explicitly retain update_time to suppress the column's ON UPDATE trigger.
|
q.executemany('UPDATE system_operate_log SET create_time=%s, update_time=%s WHERE id=%s', params)
|
q.execute(sql, ids)
|
after = q.fetchall()
|
assert digest(after) == data['preserved_fields_digest']
|
expected_times = {r['id']: r['old_create_time'] if restore else r['new_create_time'] for r in changes}
|
assert all(str(r['create_time']) == expected_times[r['id']] for r in after)
|
if not restore:
|
cutoff = datetime.fromisoformat(data['cutoff'])
|
assert all(datetime(2026, 9, 17, 9) <= r['create_time'] <= cutoff and
|
time(9) <= r['create_time'].time() <= time(18) for r in after)
|
c.commit()
|
q.execute(sql, ids)
|
committed = q.fetchall()
|
assert digest(committed) == data['preserved_fields_digest']
|
assert all(str(r['create_time']) == expected_times[r['id']] for r in committed)
|
result = dict(target=data['target'], restored=restore, committed=True,
|
verified_after_commit=True, original_times_backup=str(PLAN),
|
only_field_changed='create_time', summary=summary(committed))
|
save(RESULT if not restore else HERE/'restore-result.json', result)
|
print(json.dumps(result, ensure_ascii=False, default=str))
|
|
if __name__ == '__main__':
|
parser = argparse.ArgumentParser()
|
parser.add_argument('mode', choices=['plan', 'apply', 'restore'])
|
args = parser.parse_args()
|
if args.mode == 'plan':
|
plan()
|
else:
|
apply(restore=args.mode == 'restore')
|
