from __future__ import annotations import base64 import json import re from datetime import datetime, timezone PROJECT_TEXT_KEYS = [ "project_name", "project_client", "project_notes", "project_itinerary", ] PROJECT_SETTING_KEYS = [ "cover_base_name", "cover_frame_name", "cover_line1", "cover_line2", "cover_text_color", "cover_stroke_color", "cover_stroke_width", "cover_font1_size", "cover_font2_size", "cover_hs1", "cover_hs2", "map_base", "map_base_style", "map_base_style_pack", "map_style", "map_lang", "map_variant", "map_arrival", "map_departure", "map_itinerary", "dp_size", "dp_frame_key", "dp_boost", "dp_site", "dp_itinerary", "dp_day_filter", ] TRANSIENT_PREFIXES = [ "dp_kw1_", "dp_kw2_", "dp_img_", "dp_pool_", "dp_urls_", ] def _encode_bytes(data: bytes | None) -> str | None: if not data: return None return base64.b64encode(data).decode("ascii") def _decode_bytes(value: str | None) -> bytes | None: if not value: return None return base64.b64decode(value.encode("ascii")) def safe_name(value: str) -> str: value = (value or "tk-sales-project").strip().lower() value = re.sub(r"[^a-z0-9]+", "-", value) return value.strip("-") or "tk-sales-project" def project_stats(session_state) -> dict: double_count = 0 for item in session_state.get("dp_days", []) or []: if session_state.get(f"dp_img_{item.get('day')}"): double_count += 1 return { "has_cover": bool(session_state.get("cover_bytes")), "has_map": bool(session_state.get("map_bytes")), "double_count": double_count, "asset_count": len(session_state.get("project_assets", []) or []), "saved_output_count": sum(len(items) for items in (session_state.get("saved_outputs", {}) or {}).values()), } def _capture_double_entries(session_state) -> list[dict]: entries = [] for item in session_state.get("dp_days", []) or []: day_num = item.get("day") if day_num is None: continue entries.append( { "day": day_num, "text": item.get("text", ""), "kw1": session_state.get(f"dp_kw1_{day_num}", ""), "kw2": session_state.get(f"dp_kw2_{day_num}", ""), "image": _encode_bytes(session_state.get(f"dp_img_{day_num}")), } ) return entries def capture_bundle(session_state) -> dict: return { "schema_version": 1, "exported_at": datetime.now(timezone.utc).isoformat(), "project": {key: session_state.get(key, "") for key in PROJECT_TEXT_KEYS}, "settings": {key: session_state.get(key) for key in PROJECT_SETTING_KEYS}, "cover": {"image": _encode_bytes(session_state.get("cover_bytes"))}, "map": { "legs": session_state.get("map_legs", []) or [], "pending": session_state.get("map_pending", []) or [], "image": _encode_bytes(session_state.get("map_bytes")), "render_version": session_state.get("map_render_version"), "custom_anchors": session_state.get("map_custom_anchors", {}) or {}, }, "double_photo": { "days": session_state.get("dp_days", []) or [], "entries": _capture_double_entries(session_state), }, "photo_assets": [ { "name": item.get("name", "asset"), "mime": item.get("mime", "image/jpeg"), "data": _encode_bytes(item.get("data")), } for item in (session_state.get("project_assets", []) or []) if item.get("data") ], "saved_outputs": { kind: [ { "name": item.get("name", "output"), "mime": item.get("mime", "application/octet-stream"), "data": _encode_bytes(item.get("data")), } for item in items if item.get("data") ] for kind, items in (session_state.get("saved_outputs", {}) or {}).items() }, } def bundle_to_bytes(session_state) -> bytes: return json.dumps(capture_bundle(session_state), ensure_ascii=False, indent=2).encode("utf-8") def clear_project_state(session_state) -> None: for key in PROJECT_TEXT_KEYS + PROJECT_SETTING_KEYS: session_state.pop(key, None) for key in ["cover_bytes", "map_legs", "map_pending", "map_bytes", "map_render_version", "map_custom_anchors", "dp_days", "project_assets", "saved_outputs", "_project_bundle_digest"]: session_state.pop(key, None) for key in list(session_state.keys()): if any(key.startswith(prefix) for prefix in TRANSIENT_PREFIXES): session_state.pop(key, None) def apply_bundle(bundle: dict, session_state) -> None: clear_project_state(session_state) for key, value in (bundle.get("project") or {}).items(): session_state[key] = value for key, value in (bundle.get("settings") or {}).items(): if value is not None: session_state[key] = value cover_image = _decode_bytes((bundle.get("cover") or {}).get("image")) if cover_image: session_state["cover_bytes"] = cover_image map_data = bundle.get("map") or {} session_state["map_legs"] = map_data.get("legs", []) or [] session_state["map_pending"] = map_data.get("pending", []) or [] session_state["map_custom_anchors"] = map_data.get("custom_anchors", {}) or {} if map_data.get("render_version") is not None: session_state["map_render_version"] = map_data.get("render_version") map_image = _decode_bytes(map_data.get("image")) if map_image: session_state["map_bytes"] = map_image dp_data = bundle.get("double_photo") or {} session_state["dp_days"] = dp_data.get("days", []) or [] for item in dp_data.get("entries", []) or []: day_num = item.get("day") if day_num is None: continue session_state[f"dp_kw1_{day_num}"] = item.get("kw1", "") session_state[f"dp_kw2_{day_num}"] = item.get("kw2", "") image_bytes = _decode_bytes(item.get("image")) if image_bytes: session_state[f"dp_img_{day_num}"] = image_bytes session_state["project_assets"] = [ { "name": item.get("name", "asset"), "mime": item.get("mime", "image/jpeg"), "data": _decode_bytes(item.get("data")), } for item in (bundle.get("photo_assets") or []) if item.get("data") ] session_state["saved_outputs"] = { kind: [ { "name": item.get("name", "output"), "mime": item.get("mime", "application/octet-stream"), "data": _decode_bytes(item.get("data")), } for item in items if item.get("data") ] for kind, items in (bundle.get("saved_outputs") or {}).items() } def restore_bundle_bytes(raw_bytes: bytes, session_state) -> dict: bundle = json.loads(raw_bytes.decode("utf-8")) apply_bundle(bundle, session_state) return bundle def bundle_download_name(session_state) -> str: return f"{safe_name(session_state.get('project_name', 'tk-sales-project'))}.json"