8e589fbc47
- Replace SQLite db module with file-based JSON storage in ~/.jigaido/
- Group bounties: ~/.jigaido/{group_id}/group.json
- User tracking: ~/.jigaido/{group_id}/user_{id}.json
- Personal bounties: ~/.jigaido/user_{id}/user.json
- Anyone can add bounties to groups; only creator can edit/delete
- Bounty IDs are sequential per group, not global
- Fix test mock compatibility issues in format_bounty function
124 lines
3.8 KiB
Python
124 lines
3.8 KiB
Python
"""Per-group JSON file storage for JIGAIDO."""
|
|
|
|
import json
|
|
import tempfile
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
DATA_DIR = Path.home() / ".jigaido"
|
|
|
|
|
|
def _ensure_dirs() -> None:
|
|
DATA_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
|
|
def group_dir_path(group_id: int) -> Path:
|
|
return DATA_DIR / f"{group_id}"
|
|
|
|
|
|
def user_personal_dir_path(user_id: int) -> Path:
|
|
return DATA_DIR / f"user_{user_id}"
|
|
|
|
|
|
def group_file_path(group_id: int) -> Path:
|
|
return group_dir_path(group_id) / "group.json"
|
|
|
|
|
|
def user_tracking_file_path(group_id: int, user_id: int) -> Path:
|
|
return group_dir_path(group_id) / f"user_{user_id}.json"
|
|
|
|
|
|
def user_personal_file_path(user_id: int) -> Path:
|
|
return user_personal_dir_path(user_id) / "user.json"
|
|
|
|
|
|
def load_group(group_id: int) -> dict:
|
|
"""Load group data from JSON file. Returns empty group structure if not found."""
|
|
_ensure_dirs()
|
|
path = group_file_path(group_id)
|
|
if not path.exists():
|
|
return {
|
|
"group_id": group_id,
|
|
"bounties": [],
|
|
}
|
|
with open(path) as f:
|
|
return json.load(f)
|
|
|
|
|
|
def save_group(group_data: dict) -> None:
|
|
"""Atomically save group data to JSON file."""
|
|
_ensure_dirs()
|
|
group_id = group_data["group_id"]
|
|
dir_path = group_dir_path(group_id)
|
|
dir_path.mkdir(parents=True, exist_ok=True)
|
|
path = group_file_path(group_id)
|
|
with tempfile.NamedTemporaryFile(mode="w", dir=dir_path, delete=False) as tmp:
|
|
json.dump(group_data, tmp, indent=2)
|
|
tmp_path = tmp.name
|
|
os.rename(tmp_path, path)
|
|
|
|
|
|
def next_bounty_id(group_data: dict) -> int:
|
|
"""Get next sequential bounty ID for group."""
|
|
existing_ids = [b["id"] for b in group_data.get("bounties", [])]
|
|
return (max(existing_ids) + 1) if existing_ids else 1
|
|
|
|
|
|
def load_user_tracking(group_id: int, user_id: int) -> dict:
|
|
"""Load user tracking data for a group. Returns empty tracking structure if not found."""
|
|
_ensure_dirs()
|
|
path = user_tracking_file_path(group_id, user_id)
|
|
if not path.exists():
|
|
return {
|
|
"user_id": user_id,
|
|
"tracked": [],
|
|
}
|
|
with open(path) as f:
|
|
return json.load(f)
|
|
|
|
|
|
def save_user_tracking(tracking_data: dict, group_id: int) -> None:
|
|
"""Atomically save user tracking data to JSON file."""
|
|
_ensure_dirs()
|
|
dir_path = group_dir_path(group_id)
|
|
dir_path.mkdir(parents=True, exist_ok=True)
|
|
path = user_tracking_file_path(group_id, tracking_data["user_id"])
|
|
with tempfile.NamedTemporaryFile(mode="w", dir=dir_path, delete=False) as tmp:
|
|
json.dump(tracking_data, tmp, indent=2)
|
|
tmp_path = tmp.name
|
|
os.rename(tmp_path, path)
|
|
|
|
|
|
def load_user_personal(user_id: int) -> dict:
|
|
"""Load user's personal bounties (DM mode). Returns empty user structure if not found."""
|
|
_ensure_dirs()
|
|
path = user_personal_file_path(user_id)
|
|
if not path.exists():
|
|
return {
|
|
"user_id": user_id,
|
|
"username": None,
|
|
"bounties": [],
|
|
}
|
|
with open(path) as f:
|
|
return json.load(f)
|
|
|
|
|
|
def save_user_personal(user_data: dict) -> None:
|
|
"""Atomically save user's personal bounties to JSON file."""
|
|
_ensure_dirs()
|
|
user_id = user_data["user_id"]
|
|
dir_path = user_personal_dir_path(user_id)
|
|
dir_path.mkdir(parents=True, exist_ok=True)
|
|
path = user_personal_file_path(user_id)
|
|
with tempfile.NamedTemporaryFile(mode="w", dir=dir_path, delete=False) as tmp:
|
|
json.dump(user_data, tmp, indent=2)
|
|
tmp_path = tmp.name
|
|
os.rename(tmp_path, path)
|
|
|
|
|
|
def next_personal_bounty_id(user_data: dict) -> int:
|
|
"""Get next sequential bounty ID for user's personal bounties."""
|
|
existing_ids = [b["id"] for b in user_data.get("bounties", [])]
|
|
return (max(existing_ids) + 1) if existing_ids else 1
|