elpatron
fbc2deec45
Prepare the UI for English (default/fallback) and German with auto or manual locale selection, and report import issues with client-translated warnings instead of failing on minor save format changes. Co-authored-by: Cursor <cursoragent@cursor.com>
422 lines
15 KiB
Python
422 lines
15 KiB
Python
"""Parse and normalize Idle Fantasy Android save files."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from categories import categorize_item
|
||
from validation import Issue, analyze_save, has_errors, issue
|
||
|
||
|
||
class SaveParseError(Exception):
|
||
"""Save cannot be imported."""
|
||
|
||
def __init__(self, message: str, issues: list[Issue] | None = None):
|
||
super().__init__(message)
|
||
self.issues = issues or []
|
||
|
||
|
||
def _maybe_parse_json(value: Any, field_path: str, failures: list[str]) -> Any:
|
||
if isinstance(value, str):
|
||
stripped = value.strip()
|
||
if stripped.startswith(("{", "[")):
|
||
try:
|
||
return json.loads(stripped)
|
||
except json.JSONDecodeError:
|
||
failures.append(field_path)
|
||
return value
|
||
return value
|
||
|
||
|
||
def _deep_parse(obj: Any, path: str = "", failures: list[str] | None = None) -> Any:
|
||
failures = failures if failures is not None else []
|
||
if isinstance(obj, dict):
|
||
return {
|
||
k: _deep_parse(_maybe_parse_json(v, f"{path}.{k}" if path else k, failures), f"{path}.{k}" if path else k, failures)
|
||
for k, v in obj.items()
|
||
}
|
||
if isinstance(obj, list):
|
||
return [
|
||
_deep_parse(_maybe_parse_json(v, f"{path}[{i}]", failures), f"{path}[{i}]", failures)
|
||
for i, v in enumerate(obj)
|
||
]
|
||
return obj
|
||
|
||
|
||
def _ensure_dict(value: Any, field: str, issues: list[Issue]) -> dict[str, Any]:
|
||
if value is None:
|
||
return {}
|
||
if isinstance(value, dict):
|
||
return value
|
||
issues.append(issue(
|
||
"warning", "coerced_empty_dict",
|
||
f'Field "{field}" is not an object – treated as empty.',
|
||
field=field,
|
||
params={"field": field},
|
||
))
|
||
return {}
|
||
|
||
|
||
def _ensure_list(value: Any, field: str, issues: list[Issue]) -> list[Any]:
|
||
if value is None:
|
||
return []
|
||
if isinstance(value, list):
|
||
return value
|
||
issues.append(issue(
|
||
"warning", "coerced_empty_list",
|
||
f'Field "{field}" is not a list – skipped.',
|
||
field=field,
|
||
params={"field": field},
|
||
))
|
||
return []
|
||
|
||
|
||
def _safe_int(value: Any, field: str, issues: list[Issue], default: int = 0) -> int:
|
||
if value is None:
|
||
return default
|
||
if isinstance(value, bool):
|
||
issues.append(issue(
|
||
"warning", "invalid_number",
|
||
f'Invalid number in "{field}".',
|
||
field=field,
|
||
params={"field": field, "detail": ""},
|
||
))
|
||
return default
|
||
try:
|
||
return int(value)
|
||
except (TypeError, ValueError):
|
||
issues.append(issue(
|
||
"warning", "invalid_number",
|
||
f'Invalid number in "{field}": {value!r}.',
|
||
field=field,
|
||
params={"field": field, "detail": f": {value!r}"},
|
||
))
|
||
return default
|
||
|
||
|
||
def xp_for_level(level: int) -> int:
|
||
if level <= 1:
|
||
return 0
|
||
total = 0
|
||
for lv in range(1, level):
|
||
total += int(lv + 300 * (2 ** (lv / 7.0)))
|
||
return total // 4
|
||
|
||
|
||
def xp_to_next_level(level: int, xp: int) -> dict[str, int | float]:
|
||
current_threshold = xp_for_level(level)
|
||
next_threshold = xp_for_level(level + 1)
|
||
span = max(next_threshold - current_threshold, 1)
|
||
progress = max(0, min(xp - current_threshold, span))
|
||
return {
|
||
"xp_in_level": progress,
|
||
"xp_needed": span,
|
||
"progress_pct": round(100 * progress / span, 1),
|
||
}
|
||
|
||
|
||
def format_item_name(key: str) -> str:
|
||
return str(key).replace("_", " ").title()
|
||
|
||
|
||
def format_key(key: str) -> str:
|
||
return str(key).replace("_", " ").title()
|
||
|
||
|
||
def load_save(path: str | Path) -> tuple[dict[str, Any], list[str]]:
|
||
path = Path(path)
|
||
try:
|
||
with path.open(encoding="utf-8") as f:
|
||
raw = json.load(f)
|
||
except json.JSONDecodeError as exc:
|
||
raise SaveParseError(f"Invalid JSON file: {exc}") from exc
|
||
except OSError as exc:
|
||
raise SaveParseError(f"Could not read file: {exc}") from exc
|
||
|
||
failures: list[str] = []
|
||
parsed = _deep_parse(raw, failures=failures)
|
||
if not isinstance(parsed, dict):
|
||
raise SaveParseError("The file must contain a JSON object.")
|
||
return parsed, failures
|
||
|
||
|
||
def normalize_save(
|
||
raw: dict[str, Any],
|
||
source_file: str = "",
|
||
issues: list[Issue] | None = None,
|
||
nested_failures: list[str] | None = None,
|
||
) -> dict[str, Any]:
|
||
issues = list(issues or [])
|
||
issues.extend(analyze_save(raw, nested_failures))
|
||
|
||
if has_errors(issues):
|
||
fatal = next(i for i in issues if i["level"] == "error")
|
||
raise SaveParseError(fatal["message"], issues)
|
||
|
||
flags = _ensure_dict(raw.get("flags"), "flags", issues)
|
||
skill_levels = _ensure_dict(raw.get("skillLevels"), "skillLevels", issues)
|
||
skill_xp = _ensure_dict(raw.get("skillXp"), "skillXp", issues)
|
||
inventory = _ensure_dict(raw.get("inventory"), "inventory", issues)
|
||
equipped = _ensure_dict(raw.get("equipped"), "equipped", issues)
|
||
equipped_values = {v for v in equipped.values() if v}
|
||
|
||
inventory_items = []
|
||
for key, qty_raw in sorted(inventory.items()):
|
||
qty = _safe_int(qty_raw, f"inventory.{key}", issues, default=-1)
|
||
if qty <= 0:
|
||
if qty < 0:
|
||
continue
|
||
continue
|
||
item_key = str(key)
|
||
inventory_items.append({
|
||
"key": item_key,
|
||
"name": format_item_name(item_key),
|
||
"qty": qty,
|
||
"category": categorize_item(item_key),
|
||
"equipped": item_key in equipped_values,
|
||
})
|
||
|
||
all_skill_keys = set(skill_levels) | set(skill_xp)
|
||
skills = []
|
||
total_level = 0
|
||
for key in sorted(all_skill_keys):
|
||
level = _safe_int(skill_levels.get(key, 1), f"skillLevels.{key}", issues, default=1)
|
||
xp = _safe_int(skill_xp.get(key, 0), f"skillXp.{key}", issues, default=0)
|
||
total_level += level
|
||
prog = xp_to_next_level(level, xp)
|
||
skills.append({
|
||
"key": str(key),
|
||
"name": format_key(str(key)),
|
||
"level": level,
|
||
"xp": xp,
|
||
**prog,
|
||
})
|
||
|
||
equipment = []
|
||
for slot, item_key in equipped.items():
|
||
equipment.append({
|
||
"slot": str(slot),
|
||
"slot_name": format_key(str(slot)),
|
||
"key": item_key,
|
||
"name": format_item_name(str(item_key)) if item_key else None,
|
||
})
|
||
|
||
story_quests = []
|
||
for idx, q in enumerate(_ensure_list(raw.get("questProgress"), "questProgress", issues)):
|
||
if not isinstance(q, dict):
|
||
issues.append(issue(
|
||
"warning", "invalid_quest_entry",
|
||
f"Quest entry #{idx + 1} is not an object and was skipped.",
|
||
field="questProgress",
|
||
params={"index": idx + 1},
|
||
))
|
||
continue
|
||
quest_id = q.get("questId") or q.get("id") or f"unknown_{idx}"
|
||
story_quests.append({
|
||
"id": quest_id,
|
||
"name": format_key(str(quest_id)),
|
||
"progress": _safe_int(q.get("progress"), f"questProgress[{idx}].progress", issues),
|
||
"completed": bool(q.get("completed")),
|
||
"completed_at": q.get("completedAt"),
|
||
})
|
||
|
||
daily_quests = _build_flag_quests(
|
||
flags.get("daily_quest_ids"),
|
||
flags.get("daily_quest_progress"),
|
||
flags.get("daily_quest_claimed"),
|
||
"daily_quest", issues,
|
||
)
|
||
weekly_quests = _build_flag_quests(
|
||
flags.get("weekly_quest_ids"),
|
||
flags.get("weekly_quest_progress"),
|
||
flags.get("weekly_quest_claimed"),
|
||
"weekly_quest", issues,
|
||
)
|
||
guild_quests = _build_flag_quests(
|
||
flags.get("guild_daily_ids"),
|
||
flags.get("guild_daily_progress"),
|
||
flags.get("guild_daily_claimed"),
|
||
"guild_daily", issues,
|
||
)
|
||
|
||
sessions = []
|
||
for idx, s in enumerate(_ensure_list(raw.get("sessions"), "sessions", issues)):
|
||
if not isinstance(s, dict):
|
||
issues.append(issue(
|
||
"warning", "invalid_session_entry",
|
||
f"Session entry #{idx + 1} is not an object and was skipped.",
|
||
field="sessions",
|
||
params={"index": idx + 1},
|
||
))
|
||
continue
|
||
frames = s.get("frames")
|
||
if isinstance(frames, str):
|
||
issues.append(issue(
|
||
"warning", "unparsed_session_frames",
|
||
f"Session #{idx + 1}: activity frames could not be read.",
|
||
field="sessions",
|
||
params={"index": idx + 1},
|
||
))
|
||
frames = []
|
||
elif not isinstance(frames, list):
|
||
frames = []
|
||
total_xp = sum(_safe_int(f.get("xp_gain"), f"sessions[{idx}].xp", issues) for f in frames if isinstance(f, dict))
|
||
total_kills = sum(_safe_int(f.get("kills"), f"sessions[{idx}].kills", issues) for f in frames if isinstance(f, dict))
|
||
sessions.append({
|
||
"id": s.get("session_id"),
|
||
"skill": s.get("skill_name"),
|
||
"activity": s.get("activity_key"),
|
||
"started_at": s.get("started_at"),
|
||
"ends_at": s.get("ends_at"),
|
||
"completed": s.get("completed"),
|
||
"total_xp": total_xp,
|
||
"total_kills": total_kills,
|
||
"frame_count": len(frames),
|
||
})
|
||
|
||
pets_raw = raw.get("pets")
|
||
pets: list[Any] = []
|
||
if isinstance(pets_raw, list):
|
||
pets = pets_raw
|
||
elif pets_raw is not None:
|
||
issues.append(issue("warning", "invalid_pets", 'Field "pets" is not a list.', field="pets"))
|
||
|
||
farming = []
|
||
for idx, patch in enumerate(_ensure_list(raw.get("farmingPatches"), "farmingPatches", issues)):
|
||
if isinstance(patch, dict):
|
||
farming.append(patch)
|
||
else:
|
||
issues.append(issue(
|
||
"warning", "invalid_farming_patch",
|
||
f"Farming patch #{idx + 1} was skipped.",
|
||
field="farmingPatches",
|
||
params={"index": idx + 1},
|
||
))
|
||
|
||
if not flags.get("character_name"):
|
||
issues.append(issue(
|
||
"warning", "missing_character_name",
|
||
"No character name found in save.",
|
||
field="flags.character_name",
|
||
))
|
||
|
||
# Deduplizieren (gleicher code+field+message)
|
||
seen = set()
|
||
unique_issues: list[Issue] = []
|
||
for item in issues:
|
||
key = (item["level"], item["code"], item.get("field"), item["message"])
|
||
if key not in seen:
|
||
seen.add(key)
|
||
unique_issues.append(item)
|
||
|
||
warning_count = sum(1 for i in unique_issues if i["level"] == "warning")
|
||
info_count = sum(1 for i in unique_issues if i["level"] == "info")
|
||
|
||
return {
|
||
"character": {
|
||
"name": flags.get("character_name"),
|
||
"gender": flags.get("character_gender"),
|
||
"race": flags.get("character_race"),
|
||
"hp": flags.get("current_hp"),
|
||
"active_potion": flags.get("active_potion_key"),
|
||
"active_spell": flags.get("active_spell"),
|
||
"active_weapon_slot": flags.get("active_weapon_slot"),
|
||
"active_blessing": flags.get("active_blessing_key"),
|
||
"blessing_expires_at": flags.get("active_blessing_expires_at"),
|
||
"theme": flags.get("theme_preference"),
|
||
},
|
||
"skills": skills,
|
||
"inventory": inventory_items,
|
||
"equipment": equipment,
|
||
"quests": {
|
||
"story": story_quests,
|
||
"daily": daily_quests,
|
||
"weekly": weekly_quests,
|
||
"guild": guild_quests,
|
||
},
|
||
"combat": {
|
||
"enemy_kills": _ensure_dict(flags.get("enemy_kills"), "flags.enemy_kills", issues),
|
||
"dungeon_runs": _ensure_dict(flags.get("dungeon_runs"), "flags.dungeon_runs", issues),
|
||
"slayer_task": flags.get("active_slayer_task") if isinstance(flags.get("active_slayer_task"), dict) else None,
|
||
"slayer_points": _safe_int(flags.get("slayer_points"), "flags.slayer_points", issues),
|
||
},
|
||
"guild_reputation": _ensure_dict(flags.get("guild_reputation"), "flags.guild_reputation", issues),
|
||
"pets": pets,
|
||
"farming": farming,
|
||
"farming_fertilizer": _ensure_dict(flags.get("farming_fertilizer"), "flags.farming_fertilizer", issues),
|
||
"session_queue": _ensure_list(flags.get("session_queue"), "flags.session_queue", issues),
|
||
"recent_sessions": _ensure_list(flags.get("recent_sessions"), "flags.recent_sessions", issues),
|
||
"sessions": sessions,
|
||
"town_buildings": _ensure_dict(flags.get("town_building_tiers"), "flags.town_building_tiers", issues),
|
||
"meta": {
|
||
"source_file": source_file,
|
||
"exported_at": raw.get("exported_at"),
|
||
"coins": _safe_int(raw.get("coins"), "coins", issues),
|
||
"inventory_coins": _safe_int(inventory.get("coins"), "inventory.coins", issues),
|
||
"total_level": total_level,
|
||
"item_count": len(inventory_items),
|
||
"total_items": sum(i["qty"] for i in inventory_items),
|
||
"version_code": flags.get("last_seen_version_code"),
|
||
"import_report": unique_issues,
|
||
"import_summary": {
|
||
"ok": True,
|
||
"warnings": warning_count,
|
||
"infos": info_count,
|
||
},
|
||
},
|
||
# Unbekannte Top-Level-Felder für spätere Auswertung durchreichen
|
||
"extensions": {
|
||
k: raw[k] for k in raw if k not in {
|
||
"skillLevels", "skillXp", "inventory", "equipped", "flags",
|
||
"pets", "coins", "questProgress", "farmingPatches", "sessions", "exported_at",
|
||
}
|
||
} if isinstance(raw, dict) else {},
|
||
}
|
||
|
||
|
||
def _build_flag_quests(
|
||
ids: Any,
|
||
progress: Any,
|
||
claimed: Any,
|
||
label: str,
|
||
issues: list[Issue],
|
||
) -> list[dict[str, Any]]:
|
||
id_list = ids if isinstance(ids, list) else []
|
||
if ids is not None and not isinstance(ids, list):
|
||
issues.append(issue(
|
||
"warning", "invalid_quest_ids",
|
||
f"Quest IDs ({label}) are not a list.",
|
||
field=label,
|
||
params={"label": label},
|
||
))
|
||
progress_map = progress if isinstance(progress, dict) else {}
|
||
if progress is not None and not isinstance(progress, dict):
|
||
issues.append(issue(
|
||
"warning", "invalid_quest_progress",
|
||
f"Quest progress ({label}) is not an object.",
|
||
field=label,
|
||
params={"label": label},
|
||
))
|
||
claimed_list = claimed if isinstance(claimed, list) else []
|
||
claimed_set = set(claimed_list)
|
||
|
||
result = []
|
||
for qid in id_list:
|
||
qid_str = str(qid)
|
||
result.append({
|
||
"id": qid_str,
|
||
"name": format_key(qid_str),
|
||
"progress": _safe_int(progress_map.get(qid), f"{label}.{qid_str}", issues),
|
||
"claimed": qid_str in claimed_set or qid in claimed_set,
|
||
})
|
||
return result
|
||
|
||
|
||
def parse_save_file(path: str | Path) -> tuple[dict[str, Any], list[Issue]]:
|
||
path = Path(path)
|
||
raw, failures = load_save(path)
|
||
data = normalize_save(raw, source_file=path.name, nested_failures=failures)
|
||
return data, data["meta"]["import_report"]
|