101 lines
4.4 KiB
Python
101 lines
4.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Apply per-record images and textual prices to the Ali sampling import."""
|
|
import argparse
|
|
import gzip
|
|
import hashlib
|
|
import json
|
|
import shutil
|
|
import sys
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
|
|
def run(bundle, local=False):
|
|
if local:
|
|
from test_wsgi import app
|
|
else:
|
|
from app import app
|
|
from services.cache import bump_data_version
|
|
from services.db import get_db
|
|
|
|
bundle = Path(bundle)
|
|
records = json.loads((bundle / 'manifest.json').read_text(encoding='utf-8'))['records']
|
|
ids = [int(record['source_id']) for record in records]
|
|
if ids != list(range(548, 2992)):
|
|
raise RuntimeError('导入包申请编号必须连续为548-2991')
|
|
upload = ROOT / 'static/uploads/spot_application'
|
|
upload.mkdir(parents=True, exist_ok=True)
|
|
conn = get_db()
|
|
copied = []
|
|
old_paths = []
|
|
backup_rows = []
|
|
try:
|
|
with conn.cursor() as cur:
|
|
for record in records:
|
|
app_id = int(record['source_id'])
|
|
cur.execute("""SELECT a.image_path,c.product_unit_price
|
|
FROM spot_applications a INNER JOIN spot_confirmations c ON c.application_id=a.id
|
|
WHERE a.id=%s AND a.status='通过'""", (app_id,))
|
|
old = cur.fetchone()
|
|
if not old:
|
|
raise RuntimeError(f'申请编号{app_id}不存在')
|
|
backup_rows.append({'id': app_id, **old})
|
|
old_paths.append(old.get('image_path') or '')
|
|
backup = ROOT / 'var/manual-updates' / ('spot-ali-text-images-' + datetime.now().strftime('%Y%m%dT%H%M%S') + '.json.gz')
|
|
backup.parent.mkdir(parents=True, exist_ok=True)
|
|
with gzip.open(backup, 'wt', encoding='utf-8') as handle:
|
|
json.dump(backup_rows, handle, ensure_ascii=False, default=str)
|
|
with conn.cursor() as cur:
|
|
cur.execute('ALTER TABLE spot_confirmations MODIFY COLUMN product_unit_price VARCHAR(1000) NULL')
|
|
for record in records:
|
|
app_id = int(record['source_id'])
|
|
image_path = record['application']['image_path']
|
|
source = bundle / 'images' / Path(image_path).name
|
|
digest = hashlib.sha256(source.read_bytes()).hexdigest()
|
|
if digest != record['sha256']:
|
|
raise RuntimeError(f'图片校验失败：{source.name}')
|
|
target = upload / source.name
|
|
if target.exists():
|
|
if hashlib.sha256(target.read_bytes()).hexdigest() != digest:
|
|
raise RuntimeError(f'系统图片同名但内容不同：{target.name}')
|
|
else:
|
|
shutil.copyfile(source, target)
|
|
copied.append(target)
|
|
cur.execute('UPDATE spot_applications SET image_path=%s WHERE id=%s', (image_path, app_id))
|
|
cur.execute('UPDATE spot_confirmations SET product_unit_price=%s WHERE application_id=%s',
|
|
(record['confirmation'].get('product_unit_price'), app_id))
|
|
conn.commit()
|
|
referenced = set()
|
|
with conn.cursor() as cur:
|
|
cur.execute('SELECT image_path FROM spot_applications WHERE image_path IS NOT NULL AND image_path<>%s', ('',))
|
|
referenced = {row['image_path'] for row in cur.fetchall()}
|
|
for old_path in set(old_paths):
|
|
if old_path and old_path not in referenced and Path(old_path).name.startswith('named_'):
|
|
path = ROOT / old_path.lstrip('/')
|
|
if path.is_file():
|
|
path.unlink()
|
|
with app.test_request_context():
|
|
bump_data_version()
|
|
print(json.dumps({'updated': len(records), 'text_prices': sum(bool(r['confirmation'].get('product_unit_price')) for r in records),
|
|
'distinct_image_paths': len({r['application']['image_path'] for r in records}),
|
|
'backup': str(backup.relative_to(ROOT))}, ensure_ascii=False))
|
|
except Exception:
|
|
conn.rollback()
|
|
for path in copied:
|
|
try: path.unlink()
|
|
except OSError: pass
|
|
raise
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument('--bundle', required=True)
|
|
parser.add_argument('--local', action='store_true')
|
|
args = parser.parse_args()
|
|
run(args.bundle, args.local)
|