79 lines
2.2 KiB
Python
79 lines
2.2 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""Merge new questions into an existing question file, deduplicating by ID.
|
||
|
|
|
||
|
|
Usage:
|
||
|
|
python3 merge_questions.py <existing_file> <new_file> <output_file>
|
||
|
|
|
||
|
|
If existing_file does not exist or is empty, treats it as having no questions.
|
||
|
|
Outputs a JSON summary to stdout with counts.
|
||
|
|
|
||
|
|
Exit codes: 0 = success, 1 = error, 2 = usage error.
|
||
|
|
"""
|
||
|
|
import json
|
||
|
|
import sys
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
|
||
|
|
def load_json(path: str) -> dict | None:
|
||
|
|
"""Load JSON file, return None if missing or empty."""
|
||
|
|
p = Path(path)
|
||
|
|
if not p.exists() or p.stat().st_size == 0:
|
||
|
|
return None
|
||
|
|
with open(p, encoding="utf-8") as f:
|
||
|
|
return json.load(f)
|
||
|
|
|
||
|
|
|
||
|
|
def merge(existing: dict | None, new_data: dict) -> dict:
|
||
|
|
"""Merge new questions into existing, deduplicating by ID."""
|
||
|
|
existing_questions = (existing or {}).get("questions", [])
|
||
|
|
new_questions = new_data.get("questions", [])
|
||
|
|
|
||
|
|
existing_ids = {q["id"] for q in existing_questions}
|
||
|
|
added = [q for q in new_questions if q["id"] not in existing_ids]
|
||
|
|
skipped = len(new_questions) - len(added)
|
||
|
|
merged = existing_questions + added
|
||
|
|
|
||
|
|
# Use new_data as base, replace questions with merged
|
||
|
|
result = dict(new_data)
|
||
|
|
result["questions"] = merged
|
||
|
|
|
||
|
|
return result, {
|
||
|
|
"existing_count": len(existing_questions),
|
||
|
|
"new_count": len(new_questions),
|
||
|
|
"added_count": len(added),
|
||
|
|
"skipped_count": skipped,
|
||
|
|
"total_count": len(merged),
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def main():
|
||
|
|
if len(sys.argv) != 4:
|
||
|
|
print(f"Usage: {sys.argv[0]} <existing_file> <new_file> <output_file>",
|
||
|
|
file=sys.stderr)
|
||
|
|
sys.exit(2)
|
||
|
|
|
||
|
|
existing_path = sys.argv[1]
|
||
|
|
new_path = sys.argv[2]
|
||
|
|
output_path = sys.argv[3]
|
||
|
|
|
||
|
|
existing = load_json(existing_path)
|
||
|
|
|
||
|
|
try:
|
||
|
|
with open(new_path, encoding="utf-8") as f:
|
||
|
|
new_data = json.load(f)
|
||
|
|
except (FileNotFoundError, json.JSONDecodeError) as e:
|
||
|
|
print(f"❌ Cannot read new questions file: {e}", file=sys.stderr)
|
||
|
|
sys.exit(1)
|
||
|
|
|
||
|
|
merged, stats = merge(existing, new_data)
|
||
|
|
|
||
|
|
with open(output_path, "w", encoding="utf-8") as f:
|
||
|
|
json.dump(merged, f, ensure_ascii=False, indent=2)
|
||
|
|
|
||
|
|
print(json.dumps(stats, ensure_ascii=False))
|
||
|
|
sys.exit(0)
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
main()
|