update symbology on agol feature layer view

StuartMoore

has anyone managed to update the symbology on an agol hosted feature view layer?

i have tried the following code to update the json and it says it has worked but it dosent change if i view the layer in a map...

here's the code:

def search_layer(conn,layer_name):
    search_results = conn.content.search(layer_name, item_type='Feature Layer')
    proper_index = [i for i, s in enumerate(search_results) if 
                    '"'+layer_name+'"' in str(s)]
    found_item = search_results[proper_index[0]]
    flc = FeatureLayerCollection.fromitem(found_item)
    return flc


def update_layer_def(layer):
    # Open JSON file containing symbology update
    with open(r'C:\Temp\rp_sym.json') as json_data:
        data = json.load(json_data)
    print(data)
    layer.manager.update_definition(data)
    print("*******************UPDATED DEFINITION**********************")
    print(layer.properties)

    
def main():
    conn = GIS("pro")
    
    # Search for item, get item data)
    flc = search_layer(conn, 'view SM')
    # print(flc)
    # for x in flc.layers:
    #     print(x.properties.name)
    #     print(x.properties.id)
    layer = flc.layers[12]
    print(layer.properties.name)
    update_layer_def(layer)

if __name__ == '__main__':
    sys.exit(main())

StuartMoore

got it working, my original aim was to pull the symbology from a map and update the view from that, if anyone is interested here's the code (created with help from Copilot...):

# -*- coding: utf-8 -*-
"""
Sync symbology: Web Map (including Group Layers) → Hosted Feature Service (all sublayers)
- Recursively traverses group layers in the web map
- Matches by URL, then robust/fuzzy Name, then layerId
- Updates Hosted Feature Layer via updateDefinition
- Updates Hosted View Layer via item-data update (by layerId)
- Optional DRY_RUN and CSV audit

Run inside ArcGIS Pro (uses GIS("pro")) with ArcGIS API for Python.
"""

from arcgis.gis import GIS
from arcgis.features import FeatureLayerCollection
import sys
import re
import csv
from datetime import datetime


# ----------------------------
# Configuration
# ----------------------------
WEBMAP_ITEM_ID = "9b7...."
FEATURE_SERVICE_ITEM_ID = "010..."

# If True, only take symbology from webmap layers that point to this feature service (by URL)
FILTER_TO_TARGET_SERVICE = False

# If True, show what WOULD be updated, but do not write changes
DRY_RUN = False

# Optional: write a CSV audit of results (set to None to disable)
CSV_AUDIT_PATH = r"C:\Temp\webmap_to_fs_symbology_audit.csv"  # or None


# ----------------------------
# Helpers: normalisation & matching
# ----------------------------
def is_view_item(item) -> bool:
    """Detect hosted view layers via typeKeywords."""
    tks = set(item.typeKeywords or [])
    return "Hosted View Layer" in tks or "View Service" in tks


def norm_url(u: str) -> str:
    """Normalise URLs for comparison: lower-case, strip trailing slash."""
    if not u:
        return ""
    u = u.strip().lower()
    u = u.rstrip("/")
    return u


def same_service(node_url: str, fs_url: str) -> bool:
    """
    Check whether a web map layer URL belongs to the target feature service.
    Accept base or specific sublayer:
      node_url: .../FeatureServer    or .../FeatureServer/25
      fs_url:   .../FeatureServer
    """
    node_url = norm_url(node_url)
    fs_url = norm_url(fs_url)
    if not node_url or not fs_url:
        return False
    return node_url == fs_url or node_url.startswith(fs_url + "/")


def clean_name(s: str) -> str:
    """
    Strong normalisation for matching names.
    Handles group prefixes (/, \), punctuation, and spacing.
    """
    if not s:
        return ""
    s = s.lower().strip()

    # Remove group prefixes like "Group/Sub/Name" → "Name"
    if "/" in s:
        s = s.split("/")[-1].strip()
    if "\\" in s:
        s = s.split("\\")[-1].strip()

    # Replace common separators with space
    s = re.sub(r"[-_]+", " ", s)

    # Remove all non-alphanumeric (keep space)
    s = re.sub(r"[^a-z0-9 ]+", "", s)

    # Collapse multiple spaces
    s = " ".join(s.split())

    return s


# ----------------------------
# Extract symbology from web map (recursive)
# ----------------------------
def extract_symbology_from_node(
    node: dict,
    out_by_name: dict,
    out_by_id: dict,
    out_by_url: dict,
    target_fs_url: str = None,
    filter_to_target: bool = False,
    path: str = ""
):
    """
    Recursively extract drawingInfo from a web map node.
    Collect into:
      - out_by_name: {raw webmap name/title -> drawingInfo}
      - out_by_id:   {layerId (int)           -> drawingInfo}
      - out_by_url:  {normalized URL          -> drawingInfo}
    """
    if not isinstance(node, dict):
        return

    # If Group Layer, recurse into children
    if node.get("layerType") == "GroupLayer" and isinstance(node.get("layers"), list):
        grp_title = node.get("title") or node.get("id") or "Group"
        for child in node["layers"]:
            extract_symbology_from_node(
                child,
                out_by_name,
                out_by_id,
                out_by_url,
                target_fs_url=target_fs_url,
                filter_to_target=filter_to_target,
                path=f"{path}/{grp_title}"
            )
        return

    # Leaf node: inspect layerDefinition.drawingInfo
    ld = (node.get("layerDefinition") or {})
    di = ld.get("drawingInfo")
    if not isinstance(di, dict):
        return  # No symbology here (tables or unsupported nodes)

    # Optional filter: only include symbology for our feature service
    nurl = node.get("url")
    if filter_to_target and target_fs_url:
        if not same_service(nurl or "", target_fs_url):
            return

    # Collect by raw name/title
    raw_name = node.get("title") or node.get("id") or ""
    if raw_name:
        out_by_name[raw_name] = di

    # Collect by layerId (if present)
    if "layerId" in node and isinstance(node["layerId"], int):
        out_by_id[node["layerId"]] = di

    # Collect by URL (exact sublayer URL if present)
    if nurl:
        out_by_url[norm_url(nurl)] = di


def get_webmap_symbology(webmap_item, target_fs_url: str = None, filter_to_target: bool = False):
    """
    Walk operationalLayers and return dicts:
      - by_name_raw: {raw webmap name/title -> drawingInfo}
      - by_id:       {layerId (int)         -> drawingInfo}
      - by_url:      {normalized URL        -> drawingInfo}
    """
    wm_data = webmap_item.get_data()
    if not wm_data or "operationalLayers" not in wm_data:
        raise RuntimeError("Web map has no operationalLayers.")

    by_name_raw = {}
    by_id = {}
    by_url = {}

    for ol in wm_data["operationalLayers"]:
        extract_symbology_from_node(
            ol,
            out_by_name=by_name_raw,
            out_by_id=by_id,
            out_by_url=by_url,
            target_fs_url=target_fs_url,
            filter_to_target=filter_to_target,
            path=""
        )

    print(f"✅ Extracted symbology from web map: {len(by_name_raw)} by name, {len(by_id)} by layerId, {len(by_url)} by URL.")
    return by_name_raw, by_id, by_url


# ----------------------------
# Update functions
# ----------------------------
def update_feature_layer_sym(layer, drawingInfo, dry_run=False):
    """UpdateDefinition for normal hosted feature layer."""
    payload = {"drawingInfo": drawingInfo}
    if dry_run:
        print(f"  → [DRY_RUN] Would update '{layer.properties.name}' via updateDefinition")
        return {"dry_run": True}
    print(f"  → Updating '{layer.properties.name}' via updateDefinition...")
    res = layer.manager.update_definition(payload)
    print("    updateDefinition response:", res)
    return res


def update_view_layer_item_data_by_layer_id(view_item, layer_id: int, drawingInfo: dict, dry_run=False):
    """
    For Hosted View Layers: update item_data.layers[*] where id == layer_id,
    setting layerDefinition.drawingInfo = drawingInfo. Creates the entry if absent.
    """
    data = view_item.get_data() or {}
    if "layers" not in data or not isinstance(data["layers"], list):
        data["layers"] = []

    # Find index by matching the id
    target_idx = None
    for i, e in enumerate(data["layers"]):
        if isinstance(e, dict) and e.get("id") == layer_id:
            target_idx = i
            break

    if target_idx is None:
        # Append a new layer definition stub with the right id
        target_idx = len(data["layers"])
        data["layers"].append({"id": layer_id, "layerDefinition": {}})

    ld = data["layers"][target_idx].get("layerDefinition", {})
    ld["drawingInfo"] = drawingInfo
    data["layers"][target_idx]["layerDefinition"] = ld

    if dry_run:
        print(f"  → [DRY_RUN] Would update view layer item_data for layerId {layer_id}")
        return True

    print(f"  → Updating view layer item_data for layerId {layer_id}...")
    ok = view_item.update(data=data)
    print("    item.update response:", ok)
    return ok


# ----------------------------
# Main
# ----------------------------
def main():
    gis = GIS("pro")

    # Load web map
    wm_item = gis.content.get(WEBMAP_ITEM_ID)
    if wm_item is None:
        raise ValueError(f"Web map item '{WEBMAP_ITEM_ID}' not found.")
    print(f"Loaded web map: {wm_item.title}")

    # Load feature service
    fs_item = gis.content.get(FEATURE_SERVICE_ITEM_ID)
    if fs_item is None:
        raise ValueError(f"Feature service item '{FEATURE_SERVICE_ITEM_ID}' not found.")
    print(f"Loaded feature service: {fs_item.title}")

    fs_url = (fs_item.url or "").rstrip("/")
    flc = FeatureLayerCollection.fromitem(fs_item)
    layers = flc.layers or []
    print(f"Found {len(layers)} sublayers in the feature service.")

    # Extract symbology from the web map (recursively), optionally filtering to target FS
    by_name_raw, by_id, by_url = get_webmap_symbology(
        wm_item,
        target_fs_url=fs_url,
        filter_to_target=FILTER_TO_TARGET_SERVICE
    )

    # Precompute cleaned name mapping for web map names (for fuzzy matching)
    # Store mapping: cleaned_name -> list of (raw_name, drawingInfo)
    wm_clean_to_entries = {}
    for raw_name, di in by_name_raw.items():
        c = clean_name(raw_name)
        wm_clean_to_entries.setdefault(c, []).append((raw_name, di))

    # Determine update mode (view vs feature)
    view_mode = is_view_item(fs_item)
    print("Detected Hosted View Layer" if view_mode else "Detected Hosted Feature Layer")
    if DRY_RUN:
        print("⚐ DRY_RUN is ON — no changes will be written")

    updated = 0
    missing = []
    rows_for_csv = []

    # Iterate all sublayers of the feature service
    for lyr in layers:
        lyr_name = lyr.properties.name
        lyr_id = lyr.properties.id  # int
        fs_layer_url = f"{fs_url}/{lyr_id}"
        fs_layer_url_norm = norm_url(fs_layer_url)
        fs_name_clean = clean_name(lyr_name)

        match_strategy = None
        drawingInfo = None
        wm_match_label = ""

        # --- Strategy 1: URL match (most robust) ---
        if fs_layer_url_norm in by_url:
            drawingInfo = by_url[fs_layer_url_norm]
            match_strategy = "url"
            wm_match_label = fs_layer_url_norm

        # --- Strategy 2: Exact cleaned name match ---
        if drawingInfo is None:
            if fs_name_clean in wm_clean_to_entries:
                # Take the first with exact cleaned match
                raw_name, di = wm_clean_to_entries[fs_name_clean][0]
                drawingInfo = di
                match_strategy = "name_exact"
                wm_match_label = raw_name

        # --- Strategy 3: Fuzzy contains (FS in WM) ---
        if drawingInfo is None:
            for wm_clean, pairs in wm_clean_to_entries.items():
                if fs_name_clean and fs_name_clean in wm_clean:
                    raw_name, di = pairs[0]
                    drawingInfo = di
                    match_strategy = "name_contains_fs_in_wm"
                    wm_match_label = raw_name
                    break

        # --- Strategy 4: Fuzzy contains (WM in FS) ---
        if drawingInfo is None:
            for wm_clean, pairs in wm_clean_to_entries.items():
                if wm_clean and wm_clean in fs_name_clean:
                    raw_name, di = pairs[0]
                    drawingInfo = di
                    match_strategy = "name_contains_wm_in_fs"
                    wm_match_label = raw_name
                    break

        # --- Strategy 5: layerId fallback ---
        if drawingInfo is None and isinstance(lyr_id, int) and lyr_id in by_id:
            drawingInfo = by_id[lyr_id]
            match_strategy = "layer_id"
            wm_match_label = f"layerId={lyr_id}"

        # Apply update or log miss
        if drawingInfo is None:
            print(f"⚠ No symbology match for FS layer [{lyr_id}] '{lyr_name}'")
            missing.append((lyr_id, lyr_name))
            rows_for_csv.append({
                "timestamp": datetime.utcnow().isoformat(),
                "fs_title": fs_item.title,
                "fs_layer_id": lyr_id,
                "fs_layer_name": lyr_name,
                "matched": False,
                "strategy": "",
                "webmap_match": ""
            })
            continue

        print(f"\n=== Updating FS layer [{lyr_id}] '{lyr_name}' (match: {match_strategy} → {wm_match_label}) ===")

        if view_mode:
            update_view_layer_item_data_by_layer_id(fs_item, lyr_id, drawingInfo, dry_run=DRY_RUN)
        else:
            update_feature_layer_sym(lyr, drawingInfo, dry_run=DRY_RUN)

        updated += 1
        rows_for_csv.append({
            "timestamp": datetime.utcnow().isoformat(),
            "fs_title": fs_item.title,
            "fs_layer_id": lyr_id,
            "fs_layer_name": lyr_name,
            "matched": True,
            "strategy": match_strategy,
            "webmap_match": wm_match_label
        })

    # Summary
    print("\n✅ COMPLETE")
    print(f"Updated {updated} of {len(layers)} sublayers.")
    if missing:
        print("No symbology found for:")
        for lid, lname in missing:
            print(f"  - id {lid}, name '{lname}'")

    # Optional CSV audit
    if CSV_AUDIT_PATH:
        fieldnames = ["timestamp", "fs_title", "fs_layer_id", "fs_layer_name", "matched", "strategy", "webmap_match"]
        try:
            with open(CSV_AUDIT_PATH, "w", newline="", encoding="utf-8") as f:
                writer = csv.DictWriter(f, fieldnames=fieldnames)
                writer.writeheader()
                for r in rows_for_csv:
                    writer.writerow(r)
            print(f"\n🧾 Wrote audit CSV to: {CSV_AUDIT_PATH}")
        except Exception as ex:
            print(f"⚠ Could not write CSV audit: {ex}")

    return 0


if __name__ == "__main__":
    sys.exit(main())