feat(templates): refonte complète des templates façon Notion + vues/agents-skills
Templates (v7.71.x) : - registre unifié \ emplates\ (migrations 48-49) + TemplateService.instantiate unique (UI, API v2, agent, scheduler) - sélecteur (pilule page vide, menu •••, commande /template), gestionnaire /templates, menu New ▾, From template, base inline dans un document - 141 presets système (59 pages, 42 bases, 15 blocs, 25 lignes), titre auto depuis le template, variables title réservée - récurrences RRULE + scheduler dédupliqué, agent apply_template/list_templates, API /api/templates + /api/v2/fd-templates - correctifs : bouton Templates, centrage fenêtre, filtres CSP, flux de création, variable title - tests : tests/test_fd_templates.py (19) et e2e/templates_picker.spec.js (8) Inclut le travail déjà présent dans le working tree (vues Notion : view_query/view_aggregate/form_projection/geocoding, property_types, database_table, docs agents-skills) et ignore .playwright-mcp/.
This commit is contained in:
@@ -0,0 +1,166 @@
|
||||
"""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)
|
||||
Reference in New Issue
Block a user