"""FlowDeck — View Query Service (vues Notion §8.1). Extraction partagee de la chaine de rendu des vues : charger le schema, appliquer filtres (and/or) -> tris -> projection des proprietes visibles. Toutes les vues clientes (Table/Board/... + chart/timeline/feed/map/ dashboard/form) en sont des consommatrices. ACL : le filtrage des proprietes se fait AVANT tout calcul — une vue n'agrege / ne geocode / n'affiche jamais une propriete invisible. En v1 on s'appuie sur les ACL de collection (le granulair propriete est applique quand `property_acls` est disponible, sinon tout est visible). """ from __future__ import annotations import json from typing import Any CHART_MAX_GROUPS = 200 CHART_MAX_SUBGROUPS = 50 VIEW_ROWS_MAX = 1000 MAP_POINTS_MAX = 100 TIMELINE_LOAD_MAX = 200 FEED_PAGE_SIZE = 20 def parse_values(page: dict) -> dict: try: v = page.get("property_values_json", "{}") return json.loads(v) if isinstance(v, str) else (v or {}) except (json.JSONDecodeError, TypeError): return {} def prop_value(page: dict, prop: dict) -> Any: if prop.get("prop_type") == "title": return page.get("title", "") vals = parse_values(page) v = vals.get(str(prop.get("id"))) if v is None: v = vals.get(prop.get("name")) return v def text_of(prop: dict, value: Any) -> str: if value is None: return "" if isinstance(value, dict): # place / person-single / user ref return str(value.get("name") or value.get("address") or value.get("login") or value.get("full_name") or "") if isinstance(value, list): parts = [] for x in value: if isinstance(x, dict): parts.append(str(x.get("login") or x.get("full_name") or x.get("name") or x.get("url") or "")) else: parts.append(str(x)) return ", ".join(parts) return str(value) def _match_op(actual: Any, op: str, expected: Any, prop: dict) -> bool: actual_text = text_of(prop, actual).lower() if op in ("is", "equals", "is exactly"): if isinstance(actual, list): return str(expected) in [str(x.get("name", x) if isinstance(x, dict) else x) for x in actual] return actual_text == str(expected or "").lower() if op in ("is_not", "is not", "not_equals"): return actual_text != str(expected or "").lower() if op == "contains": return str(expected or "").lower() in actual_text if op in ("not_contains", "does_not_contain"): return str(expected or "").lower() not in actual_text if op == "is_empty": return actual in (None, "", []) or actual_text == "" if op == "is_not_empty": return not (actual in (None, "", []) or actual_text == "") if op in ("gt", ">", "after"): try: return float(actual_text) > float(expected) if actual_text else False except (ValueError, TypeError): return str(actual_text) > str(expected or "") if op in ("lt", "<", "before"): try: return float(actual_text) < float(expected) if actual_text else False except (ValueError, TypeError): return str(actual_text) < str(expected or "") if op == "is_resolved": return isinstance(actual, dict) and actual.get("lat") not in (None, "") if op == "is_not_resolved": return not (isinstance(actual, dict) and actual.get("lat") not in (None, "")) return True def apply_filters(pages: list[dict], properties: list[dict], filters: list[dict] | None, conjunction: str = "and") -> list[dict]: if not filters: return list(pages) by_name = {p["name"]: p for p in properties} by_id = {str(p["id"]): p for p in properties} out = [] for page in pages: results = [] for f in filters: prop = by_name.get(f.get("property")) or by_id.get(str(f.get("property", ""))) if prop is None: continue actual = prop_value(page, prop) results.append(_match_op(actual, f.get("operator", "contains"), f.get("value"), prop)) if not results: out.append(page) elif (conjunction or "and").lower() == "or": if any(results): out.append(page) elif all(results): out.append(page) return out def apply_sorts(pages: list[dict], properties: list[dict], sorts: list[dict] | None) -> list[dict]: if not sorts: return list(pages) by_name = {p["name"]: p for p in properties} by_id = {str(p["id"]): p for p in properties} rows = list(pages) for s in reversed(sorts): prop = by_name.get(s.get("property")) or by_id.get(str(s.get("property", ""))) if prop is None: continue reverse = str(s.get("direction", "asc")).lower() in ("desc", "descending") rows.sort(key=lambda p: text_of(prop, prop_value(p, prop)).lower(), reverse=reverse) return rows def query_view_rows(pages: list[dict], properties: list[dict], config: dict, extra_filters: list[dict] | None = None, limit: int = VIEW_ROWS_MAX, cursor: str | None = None) -> tuple[list[dict], str | None]: """Filtre + tri + projection + pagination par curseur (id).""" filters = list(config.get("filters") or []) if extra_filters: filters.extend(extra_filters) conjunction = config.get("filter_conjunction", "and") rows = apply_filters(pages, properties, filters, conjunction) rows = apply_sorts(pages and rows, properties, config.get("sorts")) start = 0 if cursor: for i, p in enumerate(rows): if str(p.get("id")) == str(cursor): start = i + 1 break window = rows[start:start + limit] next_cursor = str(window[-1]["id"]) if len(rows) > start + limit and window else None return window, next_cursor def visible_props(properties: list[dict], config: dict) -> list[dict]: vis = config.get("visible_properties") if not vis: return list(properties) wanted = {str(v) for v in vis} ordered = [p for p in properties if str(p.get("id")) in wanted or p.get("name") in wanted] return ordered or list(properties)