Util

def cast_geom_to_multi(geom: str) -> str:
    """Cast a single geometry to its multi variant."""
    ogr_geom = ogr.CreateGeometryFromWkt(geom)
    geom_type_name = ogr.GeometryTypeToName(ogr_geom.GetGeometryType())
    match geom_type_name:
        case "Polygon" | "Curve Polygon":
            ogr_geom = ogr.ForceToMultiPolygon(ogr_geom)
        case "Line String" | "Circular String" | "Compound Curve":
            ogr_geom = ogr.ForceToMultiLineString(ogr_geom)
        case "Point":
            ogr_geom = ogr.ForceToMultiPoint(ogr_geom)
    wkt = ogr_geom.ExportToWkt()
    ogr_geom = None
    return wkt

def cast_geom_to_single(geom: str) -> str:
    """Cast a multi geometry to its single variant."""
    ogr_geom = ogr.CreateGeometryFromWkt(geom)
    geom_type_name = ogr.GeometryTypeToName(ogr_geom.GetGeometryType())
    match geom_type_name:
        case "Multi Polygon" | "Multi Surface":
            ogr_geom = ogr.ForceToPolygon(ogr_geom)
        case "Multi Line String" | "Multi Curve":
            ogr_geom = ogr.ForceToLineString(ogr_geom)
        case "Multi Point":
            if ogr_geom.GetGeometryCount() == 1:
                ogr_geom = ogr_geom.GetGeometryRef(0)
    wkt = ogr_geom.ExportToWkt()
    ogr_geom = None
    return wkt

def enrich_attr_tuple(
    obj: "BaseFeature", art_tuple: tuple
) -> tuple[tuple, "BaseFeature", "PropertyInfo"]:
    """Return feature property information for construction of xpath expression."""
    result = []

    current_obj = obj

    # iterate in pairs: (attr, index)
    for attr, idx in zip(art_tuple[::2], art_tuple[1::2]):
        result.extend([attr, idx])
        property_info = current_obj.get_property_info(attr)

        if property_info.list:
            if idx is None:
                current_obj = getattr(current_obj, attr)[0]
            else:
                current_obj = getattr(current_obj, attr)[idx]
        else:
            current_obj = getattr(current_obj, attr)
        if isinstance(property_info.typename, list):
            current_type = current_obj.get_name()
        else:
            current_type = property_info.typename

        if property_info.stereotype == "DataType":
            result.extend([current_type, None])

    return tuple(result), current_obj, property_info

def format_srs(srid: int, fmt: Literal["short", "url"] = "url") -> str:
    """Formats an EPSG SRID as an SRS string.

    Args:
        srid: The EPSG SRID.
        fmt: ``"short"`` for ``EPSG:<code>`` or ``"url"`` for the OGC URL form.

    Returns:
        The formatted SRS string.
    """
    sr = osr.SpatialReference()
    sr.ImportFromEPSG(int(srid))
    auth, code = sr.GetAuthorityName(None), sr.GetAuthorityCode(None)
    match fmt:
        case "short":
            return f"{auth}:{code}"
        case "url":
            return f"http://www.opengis.net/def/crs/{auth}/0/{code}"

def get_envelope(geoms: list[str]) -> tuple[float]:
    """Return a BBOX for a list of geometries.

    Args:
        geoms: A list of WKT strings.

    Returns:
        tuple: The BBOX coordinates in the format min_X, max_X, min_Y, max_Y.
    """
    ogr_geom = ogr.CreateGeometryFromWkt(geoms.pop(0))
    for geom in geoms:
        ogr_geom = ogr_geom.Union(ogr.CreateGeometryFromWkt(geom))
    bbox = ogr_geom.GetEnvelope()
    ogr_geom = None
    return bbox

def get_geometry_type_from_wkt(geom: str):
    """Derives the geometry type from a WKT string."""
    # Imported lazily: util is imported during model package init (via orm),
    # so a module-level import of definitions would be circular.
    from xplan_tools.model.appschema.definitions import (
        Line,
        MultiLine,
        MultiPoint,
        MultiPolygon,
        Point,
        Polygon,
    )

    for geom_model in (Line, MultiLine, MultiPoint, Point, Polygon, MultiPolygon):
        if re.match(geom_model.model_fields["wkt"].metadata[0].pattern, geom):
            return geom_model

def linearize_geom(geom: str) -> str:
    """Returns the linearized WKT string."""
    split_geom = geom.split(";")
    return f"{split_geom[0]};{ogr.CreateGeometryFromWkt(split_geom[1]).GetLinearGeometry().ExportToWkt()}"

def parse_art_xpath(xpath: str) -> tuple:
    """Parse an xpath expression into a tuple of feature property information and corresponding indices."""
    result = []

    for segment in xpath.split("/"):
        match = _SEGMENT_PATTERN.match(segment)
        if not match:
            raise ValueError(f"Invalid segment: {segment}")

        name, index = match.groups()

        # Skip class markers like BP_KomplexeSondernutzung
        if _CLASS_PATTERN.match(name):
            continue

        result.append(name)
        result.append(max(int(index) - 1, 0) if index is not None else None)

    return tuple(result)

def parse_srs(srs: str | dict | None) -> int | None:
    """Returns the EPSG SRID for an SRS reference.

    Accepts any notation OGR's ``SetFromUserInput`` understands (``EPSG:25832``,
    ``urn:ogc:def:crs:EPSG::25832``, the OGC URL form, WKT) as well as the JSON-FG
    ``coordRefSys`` object form (``{"type": "Reference", "href": ...}``). CRS84 and
    its OGC URI/URN aliases resolve to 4326. The JSON-FG array (compound CRS) form
    is not supported. Returns ``None`` if the SRS cannot be resolved.

    Args:
        srs: The SRS reference, as a string, a JSON-FG ``coordRefSys`` object, or None.

    Returns:
        The EPSG SRID, or None if it could not be determined.
    """
    if isinstance(srs, dict):
        srs = srs.get("href")
    if not srs:
        return None
    srs = srs.strip()
    # OGC WGS 84 lon/lat alias (bare "CRS84", "OGC:CRS84", urn/URL forms): not
    # resolvable to an EPSG code via OGR, but equivalent to EPSG:4326.
    if "CRS84" in srs:
        return 4326
    sr = osr.SpatialReference()
    try:
        sr.SetFromUserInput(srs)
        code = sr.GetAuthorityCode(None)
        if not (code and code.isdigit()):
            sr.AutoIdentifyEPSG()
            code = sr.GetAuthorityCode(None)
    except RuntimeError:
        return None
    return int(code) if code and code.isdigit() else None

def parse_uuid(
    value: str | None, exact: bool = False, raise_exception: bool = False
) -> UUID | None:
    """Check if a given string contains a valid UUID."""
    lookup = _UUID_RE.fullmatch if exact else _UUID_RE.search
    match = lookup(value or "")
    if match is None:
        if raise_exception:
            raise ValueError(f"{value} is not a valid UUID")
        return None
    return UUID(match.group())

def serialize_art_xpath(t: tuple, prefix: str = "xplan") -> str:
    """Construct xpath expression from tuple of feature property information."""
    if len(t) % 2 != 0:
        raise ValueError("Tuple must contain (name, index) pairs")

    parts = []

    for i in range(0, len(t), 2):
        name = t[i]
        index = t[i + 1]

        if not isinstance(name, str):
            raise TypeError(f"Expected str at position {i}, got {type(name)}")

        name = f"{prefix}:{name}"
        if index is None:
            parts.append(name)
        elif isinstance(index, int):
            parts.append(f"{name}[{index + 1}]")
        else:
            raise TypeError(
                f"Expected int or None at position {i + 1}, got {type(index)}"
            )

    return "/".join(parts)

def serialize_style_rules(format: Literal["json", "yaml"]):
    """Serializes the style rules for XPlanung presentational objects in the selected format.

    Args:
        format: The format to serialize to.
    """
    return (
        json.dumps(RULES, indent=2)
        if format == "json"
        else yaml.dump(RULES, allow_unicode=True, sort_keys=False)
    )

def add_navigable_role(
    source_featuretype: str,
    navigable_role: str,
    target_featuretype: str,
    appschema: str,
    appschema_version: str,
    rel_direction: Literal["forward", "inverse"],
    dependent_part: Literal["source", "target"] | None = None,
) -> Select[tuple[bool]]:
    """Build a statement that idempotently registers a navigable role.

    Args:
        source_featuretype: Owning/source feature type name.
        navigable_role: Role (association) name linking source to target.
        target_featuretype: Referenced/target feature type name.
        appschema: Appschema prefix (e.g. ``"xplan"``).
        appschema_version: Appschema version (e.g. ``"6.1"``).
        rel_direction: Whether the refs edge runs ``"forward"`` or ``"inverse"``.
        dependent_part: Ownership marker; ``"target"`` means source existentially owns target,
            ``"source"`` means target existentially owns source, ``None`` for a non-ownership role.

    Returns:
        A ``Select`` yielding the function's ``BOOLEAN`` result; execute with ``scalar_one()``.
    """
    return select(
        func.add_navigable_role(
            source_featuretype,
            navigable_role,
            target_featuretype,
            appschema,
            appschema_version,
            rel_direction,
            dependent_part,
        )
    )

def coretable_delete_object_recursive(
    start_id: UUID,
    dry_run: bool = False,
) -> Select[tuple[Feature]]:
    """Build a statement that recursively deletes ``start_id`` and its safe cascade closure.

    Warning:
        Executing the returned statement performs the deletion unless ``dry_run`` is ``True``; run
        it inside a transaction. The affected rows are captured before removal, so the statement
        still yields them after the delete.

    Args:
        start_id: Root coretable id to delete.
        dry_run: When ``True``, return the closure without deleting anything.

    Returns:
        A ``Select`` yielding the deleted (or, for ``dry_run``, would-be-deleted) ORM ``Feature``
        rows; execute with ``scalars().all()``.
    """
    deleted = aliased(
        Feature,
        func.coretable_delete_object_recursive(start_id, dry_run).table_valued(
            *Feature.__table__.c
        ),
    )
    return select(deleted)

def coretable_delete_orphans_recursive() -> Select[tuple[int]]:
    """Build a statement that deletes dependent-part objects no longer owned by any whole.

    Warning:
        Executing the returned statement performs the deletion; run it inside a transaction.

    Returns:
        A ``Select`` yielding the total number of deleted rows as an ``INTEGER``; execute with
        ``scalar_one()``.
    """
    return select(func.coretable_delete_orphans_recursive())

def coretable_feature_graph(
    start_id: UUID,
    depth_limit: int = 3,
    include_forward: bool = True,
    include_backward: bool = True,
) -> Select[tuple[Feature]]:
    """Return a chainable Select of Features reachable from ``start_id`` via the role graph.

    Args:
        start_id: Root coretable id to traverse from.
        depth_limit: Maximum BFS depth; defaults to 3 = the current appschemas' maximum depth.
        include_forward: Traverse navigable roles in the source->target direction.
        include_backward: Traverse navigable roles in the target->source direction.

    Returns:
        A chainable ``Select`` yielding ORM ``Feature`` rows.
    """
    graph_ids = func.coretable_role_graph_ids(
        start_id, depth_limit, include_forward, include_backward, False
    ).table_valued("id")
    return select(Feature).where(exists().where(Feature.id == graph_ids.c.id))

def coretable_role_graph(
    start_id: UUID,
    depth_limit: int = 0,
    include_forward: bool = True,
    include_backward: bool = True,
    dependent_parts: bool = False,
) -> Select[tuple[Feature]]:
    """Build a statement returning the ``Feature`` rows reachable from ``start_id``.

    Args:
        start_id: Root coretable id to traverse from.
        depth_limit: Maximum BFS depth; ``0`` means unbounded.
        include_forward: Traverse navigable roles in the source->target direction.
        include_backward: Traverse navigable roles in the target->source direction.
        dependent_parts: Collect the transitive existential dependent-part (ownership) closure
            instead of a general traversal; both directions are followed, ignoring the include_*
            flags.

    Returns:
        A ``Select`` yielding ORM ``Feature`` rows; execute with ``scalars().all()``.
    """
    graph = aliased(
        Feature,
        func.coretable_role_graph(
            start_id, depth_limit, include_forward, include_backward, dependent_parts
        ).table_valued(*Feature.__table__.c),
    )
    return select(graph)

def coretable_role_graph_cascade(
    start_id: UUID,
    depth_limit: int = 0,
) -> Select[tuple[Feature]]:
    """Build a statement returning the ``Feature`` rows safe to cascade-delete with ``start_id``.

    The result is the dependent-part closure of ``start_id`` minus any part whose owning whole lies
    outside that closure (transitively), i.e. the objects that can be removed together with
    ``start_id`` without orphaning a part still owned from elsewhere.

    Args:
        start_id: Root coretable id whose cascade closure is collected.
        depth_limit: Maximum BFS depth; ``0`` means unbounded.

    Returns:
        A ``Select`` yielding ORM ``Feature`` rows; execute with ``scalars().all()``.
    """
    cascade = aliased(
        Feature,
        func.coretable_role_graph_cascade(start_id, depth_limit).table_valued(
            *Feature.__table__.c
        ),
    )
    return select(cascade)

def coretable_role_graph_ids(
    start_id: UUID,
    depth_limit: int = 0,
    include_forward: bool = True,
    include_backward: bool = True,
    dependent_parts: bool = False,
) -> Select[tuple[UUID]]:
    """Build a statement returning the ids reachable from ``start_id`` via navigable roles.

    Args:
        start_id: Root coretable id to traverse from.
        depth_limit: Maximum BFS depth; ``0`` means unbounded.
        include_forward: Traverse navigable roles in the source->target direction.
        include_backward: Traverse navigable roles in the target->source direction.
        dependent_parts: Collect the transitive existential dependent-part (ownership) closure
            instead of a general traversal; both directions are followed, ignoring the include_*
            flags.

    Returns:
        A ``Select`` yielding one ``UUID`` per reachable id (empty when ``start_id`` does not
        exist); execute with ``scalars().all()``.
    """
    graph_ids = func.coretable_role_graph_ids(
        start_id, depth_limit, include_forward, include_backward, dependent_parts
    ).table_valued("id")
    return select(graph_ids.c.id)

def list_navigable_roles() -> Select[tuple[str, str, str, str, str, str, str | None]]:
    """Build a statement returning every configured navigable role.

    Returns:
        A ``Select`` yielding one row per configured role with columns ``source_featuretype``,
        ``navigable_role``, ``target_featuretype``, ``appschema``, ``appschema_version``,
        ``rel_direction`` and ``dependent_part``; execute with ``all()``.
    """
    # The function projects the business columns of navigable_roles_config, dropping the
    # surrogate `id` and the `created_at` audit column that the ORM model also carries.
    role_columns = [
        column
        for column in NavigableRolesConfig.__table__.c
        if column.name not in ("id", "created_at")
    ]
    roles = func.list_navigable_roles().table_valued(*role_columns)
    return select(roles)

def list_top_level_featuretypes() -> Select[tuple[str, str, str]]:
    """Build a statement returning the top-level (containment/deletion root) feature types.

    Returns:
        A ``Select`` yielding one row per appschema/version with columns ``featuretype``,
        ``appschema`` and ``appschema_version``; execute with ``all()``.
    """
    top_level = func.list_top_level_featuretypes().table_valued(
        "featuretype",
        "appschema",
        "appschema_version",
    )
    return select(top_level)

def add_style_properties_to_feature(
    obj: "BaseFeature",
    ref_obj: "BaseFeature",
    to_text: bool = False,
    always_populate_schriftinhalt: bool = False,
) -> "BaseFeature":
    """Add styling properties to presentational objects.

    This method parses object (dientZurDarstellungVon) and property (art) references from
    presentational objects and derives styling information (stylesheetId, schriftinhalt)
    based on a set of defined rules.

    Args:
        obj: The presentational object.
        ref_obj: The object referenced by the presentational object.
        to_text: Whether to convert symbolic presentational objects to textual ones. Defaults to False.
        always_populate_schriftinhalt: Populate `schriftinhalt` even if a rule has no text template.
    """
    uom_map = {"m2": "m²", "m3": "m³", "grad": "°"}

    def parse_art(ref_obj: "BaseFeature", art: str) -> dict:
        def parse_value(value: Any) -> dict:
            if prop_info.stereotype == "Measure":
                value = value.value
            if prop_info.typename == "Boolean":
                value = str(value).lower()

            if name.startswith("Z"):
                text = toRoman(value)
            elif prop_info.stereotype == "Enumeration":
                member = prop_info.enum(value)
                text = member.token or member.alias or member.label
            elif prop_info.stereotype == "Codelist":
                text = str(value)
                if text.startswith("urn:"):
                    text = text.split(f"urn:xplan:{prop_info.typename}:")[1]
            elif prop_info.stereotype == "Measure":
                text = f"{value:n} {uom_map.get(prop_info.uom, prop_info.uom)}"
            else:
                text = value

            return {
                "value": value,
                "text": text,
            }

        xpath_input_tuple = parse_art_xpath(art)
        enriched_tuple, value, prop_info = enrich_attr_tuple(ref_obj, xpath_input_tuple)
        name = enriched_tuple[::2][-1]

        data = {
            "name": name,
            "data": parse_value(value),
            "type": prop_info.typename,
        }
        return data

    # TODO use for XPlanung v6.1 with addition attribute massstabFaktor
    # def set_scale(obj):
    #     bereich = self.root.get(str(obj.gehoertZuBereich))
    #     plan = self.root.get(str(bereich.gehoertZuPlan))
    #     default_scale = SCALES.get(plan.get_name(), 1000)
    #     actual_scale = (
    #         bereich.erstellungsMassstab or plan.erstellungsMassstab or default_scale
    #     )
    #     if obj.skalierung <= 3:
    #         obj.skalierung = float(obj.skalierung * actual_scale / 1000)

    obj = deepcopy(obj)
    logger.debug(f"Feature {obj.id}: adding style properties")
    version = obj.appschema().version
    if to_text and (old_type := obj.get_name()) == "XP_PPO":
        new_type = "XP_PTO"
        obj = obj.appschema().model_factory(new_type).model_validate(obj.model_dump())
        logger.info(f"Feature {obj.id}: converted {old_type} to {new_type}")
    # TODO use for XPlanung v6.1 with addition attribute massstabFaktor
    # if hasattr(obj, "skalierung"):
    #     set_scale(obj)

    logger.debug(
        f"parsing properties {obj.art} for referenced feature {ref_obj.get_name()} with ID {ref_obj.id}"
    )
    selectors = {}
    for art in obj.art:
        try:
            parsed_art = parse_art(ref_obj, art)
            selectors[parsed_art.pop("name")] = parsed_art
        except Exception:
            logger.error(f"Feature {obj.id}: art '{art}' could not be parsed")
    valid_rules = []
    for rule_id, rule in RULES.items():
        versioned_rule = rule["versions"].get(version, {"selector": {}})
        if isinstance(
            versioned_rule, str
        ):  # use other versioned rule referenced by string
            versioned_rule = rule["versions"][versioned_rule]
        valid = versioned_rule["selector"].keys() == selectors.keys() and (
            all(
                (
                    filter.get("value", None) == ["*"]
                    or selectors.get(attr, {}).get("data", {}).get("value", None)
                    in filter.get("value", False)
                )
                and (
                    selectors.get(attr, {}).get("type", None)
                    == filter.get("type", False)
                )
                for attr, filter in versioned_rule["selector"].items()
            )
            if versioned_rule.get("selector", None)
            else False
        )
        if valid:
            valid_rules.append(rule_id)
            texts = {attr: data["data"]["text"] for attr, data in selectors.items()}
            obj.stylesheetId = AnyUrl(
                f"https://registry.gdi-de.org/codelist/de.xleitstelle.xplanung/XP_StylesheetListe/{rule_id}"
            )
            if (text := versioned_rule.get("text", None)) and hasattr(
                obj, "schriftinhalt"
            ):
                obj.schriftinhalt = text.format(**texts)
            elif always_populate_schriftinhalt and hasattr(obj, "schriftinhalt"):
                obj.schriftinhalt = " ".join(
                    [str(data["data"]["text"]) for data in selectors.values()]
                ).strip()
    if not valid_rules:
        logger.warning(f"No rule found for feature {obj.id}")
        obj.stylesheetId = None
        if hasattr(obj, "schriftinhalt"):
            obj.schriftinhalt = " ".join(
                [str(data["data"]["text"]) for data in selectors.values()]
            ).strip()
            logger.debug(f"Feature {obj.id}: schriftinhalt set to {obj.schriftinhalt}")
        # if all(
        #     data["type"] in ["CharacterString", "Integer", "Decimal", "Length"]
        #     for data in selectors.values()
        # ):
        #     obj.stylesheetId = "81e52187-a33b-4340-9d6e-f25533e01aa3"
        #     if hasattr(obj, "schriftinhalt"):
        #         obj.schriftinhalt = " ".join(
        #             [str(data["data"]["text"]) for data in selectors.values()]
        #         ).strip()
        #         logger.debug(
        #             f"Feature {obj.id}: schriftinhalt set to {obj.schriftinhalt}"
        #         )
        # else:
        #     logger.warning(f"No rule found for feature {obj.id}")
    if len(valid_rules) > 1:
        raise ValueError(f"More than one rules valid: {', '.join(valid_rules)}")
    else:
        logger.debug(f"Feature {obj.id}: stylesheetId set to {obj.stylesheetId}")
    return obj

def save_validation_reports(
    report_dict: dict[str, ResultReport],
    output: str = "report.json",
):
    """Save validation reports to JSON files."""
    if report_dict is None:
        return

    output_path = Path(output)

    for plan_id in report_dict.keys():
        if len(output_path.parents) > 1:
            output_path.parent.mkdir(parents=True, exist_ok=True)
        report_path = str(output_path.parent / f"{output_path.stem}_{plan_id}.json")
        with open(report_path, "wb") as f:
            f.write(to_json(report_dict[plan_id].validation_report, indent=4))
        logger.info(f"Validation report saved as {report_path}")

async def xplan_validate(
    collection: BaseCollection,
    input: str = "xplan.gml",
    single_plans: bool = False,
    validator_url: str = "https://www.xplanungsplattform.de/xplan-api-validator/xvalidator/api/v1/",
) -> dict[str, ResultReport]:
    """Validate a Feature Collection with the official XPlanValidator.

    Args:
        collection: A BaseCollection instance.
        input: An optional input file name to use in the validation report.
        single_plans: Whether to validate plans in the collection individually.
        validator_url: The base URL of the XPlanValidator instance. Must have a trailing slash.
    """
    input_path = Path(input)
    report_dict: dict[str, ResultReport] = {}

    plans = (
        collection.get_single_plans(with_name=True)
        if single_plans
        else [(input_path.stem, collection)]
    )
    async with httpx2.AsyncClient(timeout=10) as client:
        for plan_name, plan in plans:
            plan_id = getattr(
                next(
                    filter(
                        lambda value: "_Plan" in value.get_name(),
                        plan.features.values(),
                    )
                ),
                "id",
            )

            headers = {
                "accept": "application/json",
                "x-filename": f"{plan_name}{input_path.suffix}",
                "content-type": "application/gml+xml",
            }
            params = {"name": "report.json"}
            report = ResultReport()
            with io.BytesIO() as buffer:
                GMLRepository(buffer).save_all(plan)
                logger.debug(
                    f"Sending validation request for plan '{plan_name}' to XPlanValidator API @ {validator_url}"
                )
                try:
                    response = await client.post(
                        validator_url + "validate",
                        headers=headers,
                        params=params,
                        content=buffer.getvalue(),
                    )
                    response.raise_for_status()
                except httpx2.HTTPError:
                    logger.exception(
                        f"Error while requesting validation for plan '{plan_name}', skipping"
                    )
                    report.add_error(
                        code=_ErrorCodes.ERR_VALIDATION_REQUEST_FAILED,
                        message=f"Error while requesting validation for plan '{plan_name}', skipping",
                        location=plan_name,
                    )
                    continue

            validation_report = from_json(response.content)
            report.evaluate_validation_report(validation_report)
            report_dict[plan_id] = report

    return report_dict