60 lines
2.9 KiB
Python
60 lines
2.9 KiB
Python
"""纠正已知初始日期为先开放、后发Offer；备份、事务更新，可重复执行。"""
|
|
import ast
|
|
import json
|
|
import sys
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
from sqlalchemy import text
|
|
from database import engine
|
|
|
|
|
|
def main():
|
|
tree = ast.parse((ROOT / "seed_data.py").read_text(encoding="utf-8-sig"))
|
|
constants = {node.targets[0].id: ast.literal_eval(node.value)
|
|
for node in tree.body if isinstance(node, ast.Assign)
|
|
and isinstance(node.targets[0], ast.Name)
|
|
and node.targets[0].id in ("employment_data", "student_data")}
|
|
seeds = {str(constants["student_data"][row["student_id"]-1]["student_no"]): row
|
|
for row in constants["employment_data"]}
|
|
snapshot = "SELECT e.*, s.student_no FROM employment e JOIN student s ON s.sid=e.student_id ORDER BY e.id"
|
|
with engine.begin() as conn:
|
|
before = [dict(row._mapping) for row in conn.execute(text(snapshot + " FOR UPDATE"))]
|
|
backup = ROOT / "artifacts" / ("employment-before-offer-order-" + datetime.now().strftime("%Y%m%d-%H%M%S-%f") + ".json")
|
|
backup.parent.mkdir(exist_ok=True)
|
|
backup.write_text(json.dumps({"database": conn.execute(text("SELECT DATABASE()")).scalar(), "rows": before}, ensure_ascii=False, indent=2, default=str), encoding="utf-8")
|
|
changed, skipped = [], []
|
|
for old in before:
|
|
seed = seeds.get(old['student_no'])
|
|
if not seed:
|
|
continue
|
|
start = datetime.fromisoformat(seed['employment_start_time'])
|
|
offer = datetime.fromisoformat(seed['offer_time'])
|
|
assert offer >= start
|
|
if (old['employment_start_time'], old['offer_time']) == (start, offer):
|
|
continue
|
|
# 只改完全匹配的旧种子日期，保留用户已编辑的记录。
|
|
if old['company_name'] != seed['company_name'] or (old['employment_start_time'], old['offer_time']) != (offer, start):
|
|
skipped.append(old['id'])
|
|
continue
|
|
conn.execute(text("UPDATE employment SET employment_start_time=:start, offer_time=:offer WHERE id=:id"), {'start':start,'offer':offer,'id':old['id']})
|
|
changed.append(old['id'])
|
|
after = [dict(row._mapping) for row in conn.execute(text(snapshot))]
|
|
assert len(before) == len(after)
|
|
for old, new in zip(before, after):
|
|
assert old['id'] == new['id']
|
|
if old['id'] not in changed:
|
|
assert old == new
|
|
else:
|
|
assert new['offer_time'] >= new['employment_start_time']
|
|
assert all(old[key] == new[key] for key in old if key not in ('employment_start_time','offer_time'))
|
|
print('Backup:', backup)
|
|
print('Updated initial record IDs:', changed)
|
|
print('Skipped customized initial record IDs:', skipped)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|