flowdeck/app/services/view_query.py

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