33 lines
1.3 KiB
Python
33 lines
1.3 KiB
Python
"""Import an explicit, private order snapshot into the chat SQLite database."""
|
|
|
|
import argparse
|
|
import json
|
|
from pathlib import Path
|
|
import sqlite3
|
|
import sys
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
from logistics_chat.order_data import import_records
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--snapshot', type=Path, required=True)
|
|
parser.add_argument('--db', type=Path, required=True)
|
|
parser.add_argument('--replace', action='store_true', help='Replace existing rows, including manual edits')
|
|
args = parser.parse_args()
|
|
if not args.snapshot.is_file() or args.snapshot.stat().st_size > 10*1024*1024:
|
|
parser.error('Snapshot must be an existing JSON file smaller than 10 MB')
|
|
records = json.loads(args.snapshot.read_text(encoding='utf-8-sig'))
|
|
if isinstance(records, dict): records = records.get('orders')
|
|
if not isinstance(records, list) or len(records) > 3000:
|
|
parser.error('Snapshot must contain a list of at most 3000 orders')
|
|
if not args.db.parent.is_dir(): parser.error('Database parent directory must exist')
|
|
with sqlite3.connect(args.db) as connection:
|
|
result = import_records(connection, records, replace=args.replace)
|
|
print(json.dumps(result))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|