53 lines
1.8 KiB
Python
53 lines
1.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Create the immutable daily suggested-quantity snapshot used by 连续建议天数."""
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
if str(ROOT) not in sys.path:
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument('--local', action='store_true', help='Use the local test database')
|
|
args = parser.parse_args()
|
|
|
|
if args.local:
|
|
from test_wsgi import app
|
|
else:
|
|
from app import app
|
|
|
|
from flask import session
|
|
from routes import planning
|
|
from services.db import query
|
|
from services.planning_snapshot import shanghai_today, snapshot_exists, store_snapshot
|
|
|
|
|
|
def main():
|
|
day = shanghai_today()
|
|
if snapshot_exists(day):
|
|
print(json.dumps({'date': day.isoformat(), 'created': False, 'rows': 0,
|
|
'message': 'Snapshot already exists; unchanged'}, ensure_ascii=False))
|
|
return
|
|
admins = query(
|
|
"SELECT id, role FROM dashboard_users "
|
|
"WHERE role IN ('admin', 'superadmin') ORDER BY id LIMIT 1"
|
|
)
|
|
if not admins:
|
|
raise RuntimeError('No administrator account available for snapshot generation')
|
|
with app.test_request_context('/internal/planning-snapshot'):
|
|
session['user_id'] = admins[0]['id']
|
|
session['role'] = admins[0]['role']
|
|
response = planning.planning_list.__wrapped__({'page': '1', 'size': '1000000'})
|
|
body = response.get_json(silent=True) or {}
|
|
if body.get('code') != 0:
|
|
raise RuntimeError(f"Planning rows unavailable: {body.get('msg') or body}")
|
|
rows = (body.get('data') or {}).get('rows') or []
|
|
result = store_snapshot(rows, day)
|
|
print(json.dumps(result, ensure_ascii=False))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|