| 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" |
|
|
