#!/usr/bin/env python3
"""
Apply AI-generated App Store Connect metadata to the resources identified by
asc_metadata_detector.

Reads two JSON files:

  --state    detector output. Authoritative allow-list of empty fields per
             locale plus resource ids (appInfoLocalizationId /
             appStoreVersionLocalizationId).
  --response raw AI inference JSON. Expected shape:
                 {"localizations": {"en-US": {"name": "...", ...}, ...}}

PATCHes ONLY fields that appear in state['empty_fields'][locale] AND are not in
SKIP_URL_FIELDS -- belt-and-suspenders double-gate against the detector losing
track of a URL field or a subsequent manual edit populating a previously empty
field.

Non-fatal by design -- mirrors set_app_store_whats_new.py's fail-open pattern.
Any unexpected error becomes a ::warning:: and exits 0 so the TestFlight job
does not fail on best-effort metadata polish.
"""

from __future__ import annotations

import argparse
import json
import re
import sys
from typing import Any

from asc_common import make_jwt, request
from metadata_constants import (
    APP_LEVEL_FIELDS,
    CHAR_LIMITS,
    SKIP_URL_FIELDS,
    VERSION_LEVEL_FIELDS,
    log,
    require_env,
    warn,
)


APP_INFO_RESOURCE = "appInfoLocalizations"
VERSION_RESOURCE = "appStoreVersionLocalizations"

# Every App Store description ends with a functional Terms of Use link and the
# app's own privacy-policy link (Apple's "no functional link to the Terms of
# Use" metadata rejection). Terms is Apple's standard EULA; the privacy URL is
# the locale's existing `privacyPolicyUrl`, which the panel sets and the
# detector reads back into `state.localizations[locale].fields`.
TERMS_URL = "https://www.apple.com/legal/internet-services/itunes/dev/stdeula/"
DESCRIPTION_LIMIT = CHAR_LIMITS["description"]
_FOOTER_SEPARATOR = "\n\n"

# Collapses any run of ASCII whitespace (including the raw control chars AI
# sometimes emits inside string literals: \n, \r, \t, and interior spaces)
# into a single space. Applied after non-strict JSON load so field values
# stay within Apple's char-limit gates and don't contain newlines that
# ASC UI renders as literal line breaks in the store listing.
_WHITESPACE_RUN = re.compile(r"\s+")


def _normalize_whitespace(value: Any) -> Any:
    """Return str with any whitespace run -> single space, trimmed. Non-str passthrough."""
    if not isinstance(value, str):
        return value
    return _WHITESPACE_RUN.sub(" ", value).strip()


def _normalize_ai_response(data: dict[str, Any]) -> dict[str, Any]:
    """Recursively collapse whitespace in every string value under
    ``data['localizations'][locale][field]``.

    Defensive against AI responses that emit literal newlines inside string
    literals (permitted by non-strict json.loads but rejected by ASC field
    validation).
    """
    locs = data.get("localizations")
    if not isinstance(locs, dict):
        return data
    for locale, fields in list(locs.items()):
        if not isinstance(fields, dict):
            continue
        for field, value in list(fields.items()):
            fields[field] = _normalize_whitespace(value)
    return data


def _load_json(path: str) -> dict[str, Any] | None:
    try:
        with open(path, "r", encoding="utf-8") as f:
            raw = f.read()
    except OSError as exc:
        warn(f"could not read {path}: {exc!r}")
        return None
    try:
        # strict=False permits unescaped control chars (e.g. raw \n, \t)
        # inside JSON string literals per RFC 8259 s.7 relaxed reading.
        # Apple's AI responses occasionally contain literal newlines that
        # a strict parser would reject, collapsing the applier to 0 writes.
        data = json.loads(raw, strict=False)
    except json.JSONDecodeError as exc:
        # Surface WHAT the file actually contained so a broken upstream
        # capture (e.g. ai-inference output wrapped in markdown fences) is
        # diagnosable from logs alone. Cap at 500 chars to stay well under
        # GitHub Actions annotation limits.
        preview = raw[:500].replace("\n", "\\n")
        warn(
            f"could not parse {path} as JSON: {exc!r}; "
            f"length={len(raw)} first 500 chars: {preview!r}"
        )
        return None
    if not isinstance(data, dict):
        warn(f"{path} did not contain a JSON object")
        return None
    return data


def _validate_field(field: str, value: Any) -> str | None:
    """Return normalized value (trimmed, within char limit) or None to drop."""
    if field in SKIP_URL_FIELDS or not isinstance(value, str):
        return None
    trimmed = value.strip()
    if not trimmed:
        return None
    limit = CHAR_LIMITS.get(field)
    if limit is not None and len(trimmed) > limit:
        warn(f"field '{field}' length {len(trimmed)} exceeds {limit}; dropping")
        return None
    return trimmed


def footer_for(privacy_url: str) -> str:
    """The exact two-line footer appended to every generated description."""
    return f"Terms of Use: {TERMS_URL}\nPrivacy Policy: {privacy_url}"


def with_footer(body: str, privacy_url: str) -> str:
    """Return ``body`` + one blank line + the footer, inside Apple's limit.

    The body is shortened on a whitespace boundary when body + footer would
    exceed the description limit; a hard cut happens only when the body has
    no whitespace at or before the budget. The footer is never shortened.
    """
    footer = footer_for(privacy_url)
    budget = DESCRIPTION_LIMIT - len(_FOOTER_SEPARATOR + footer)
    text = body.rstrip()
    if len(text) > budget:
        cut = max(text.rfind(ch, 0, budget + 1) for ch in (" ", "\n", "\t"))
        text = text[:cut] if cut > 0 else text[:budget]
        text = text.rstrip()
    return text + _FOOTER_SEPARATOR + footer


def _describe_with_footer(
    locale: str, loc_state: dict[str, Any], ver_writes: dict[str, str]
) -> None:
    """Attach the footer to a surviving generated description, in place.

    Reads the locale's existing ``privacyPolicyUrl`` from the detector state.
    An empty URL drops the description entirely (with a warning): a footer
    with an empty privacy link would ship the exact defect this guards
    against.
    """
    if "description" not in ver_writes:
        return
    privacy = ((loc_state.get("fields") or {}).get("privacyPolicyUrl") or "").strip()
    if not privacy:
        warn(
            f"{locale}: privacyPolicyUrl is empty; skipping generated description "
            "(footer requires it)"
        )
        del ver_writes["description"]
        return
    complete = with_footer(ver_writes["description"], privacy)
    footer = footer_for(privacy)
    if len(complete) > DESCRIPTION_LIMIT or not complete.endswith(footer):
        warn(
            f"{locale}: description with footer is invalid "
            f"(length {len(complete)}); dropping"
        )
        del ver_writes["description"]
        return
    ver_writes["description"] = complete


def _build_writes(
    ai_locale: dict[str, Any],
    empty_list: list[str],
    group_fields: tuple[str, ...],
) -> dict[str, str]:
    """Filter AI values for one field-group (app-level OR version-level).

    Double-gates against `empty_list` (detector said field is empty) AND
    SKIP_URL_FIELDS (never generate URLs). Validates char limits. Returns
    {field: normalized_value} for every field that survives all gates.
    """
    empty_set = set(empty_list or ())
    writes: dict[str, str] = {}
    for field in group_fields:
        if field not in empty_set or field in SKIP_URL_FIELDS or field not in ai_locale:
            continue
        normalized = _validate_field(field, ai_locale[field])
        if normalized is not None:
            writes[field] = normalized
    return writes


def _patch_localization(
    token: str, resource: str, loc_id: str, writes: dict[str, str]
) -> None:
    request(
        "PATCH",
        f"/{resource}/{loc_id}",
        token,
        json_body={
            "data": {"type": resource, "id": loc_id, "attributes": writes}
        },
    )


def _extract_asc_detail(exc: SystemExit) -> str:
    """Pull ASC's human-readable `errors[].detail` out of a SystemExit.

    `asc_common.request` formats failures as
    ``"ASC PATCH /path failed: {status}\\n{body[:2000]}"``. The body is
    ASC's standard JSON errors envelope:
    ``{"errors":[{"status":"409","code":"...","detail":"..."}]}``.
    Parse best-effort; on any failure fall back to the raw message so we
    never lose context.
    """
    raw = str(exc)
    # Take everything after the first newline (the response body portion).
    body = raw.split("\n", 1)[1] if "\n" in raw else ""
    if not body:
        return raw
    try:
        payload = json.loads(body)
        errors = payload.get("errors") if isinstance(payload, dict) else None
        if isinstance(errors, list) and errors:
            first = errors[0] or {}
            detail = first.get("detail") or ""
            code = first.get("code") or ""
            status = first.get("status") or ""
            # Prefer detail (human-readable); fall back to code/status if absent.
            if detail:
                return f"{status} {code}: {detail}".strip(": ").strip()
            if code or status:
                return f"{status} {code}".strip()
    except (json.JSONDecodeError, TypeError, AttributeError):
        pass
    return raw


def _patch_group(
    token: str,
    resource: str,
    locale: str,
    loc_id: str | None,
    writes: dict[str, str],
) -> int:
    """PATCH one (resource, locale) pair -- one request PER FIELD.

    Per-field attempts so a single ASC 409 (e.g. whatsNew rejected because
    the version is past editable state while description/keywords/
    promotionalText still accept edits) does not block the other fields.

    Returns count of fields successfully written (0 on total skip).
    """
    if not writes:
        return 0
    if not loc_id:
        if resource == VERSION_RESOURCE:
            # Detector deliberately clears version_localization_id when the
            # app store version is in a non-editable state (e.g. IN_REVIEW,
            # WAITING_FOR_REVIEW with PATCH rejection). Surface it as a
            # normal log, not a warning -- this is the intended no-op path.
            log(
                f"skipping version-level fields for {locale} "
                f"(version not in editable state): {sorted(writes)}"
            )
        else:
            warn(f"{locale}: {resource} id missing; skipping fields {sorted(writes)}")
        return 0

    written = 0
    for field in sorted(writes):
        value = writes[field]
        try:
            _patch_localization(token, resource, loc_id, {field: value})
        except SystemExit as exc:
            detail = _extract_asc_detail(exc)
            warn(
                f"{locale}: /{resource} {field} PATCH rejected by ASC: {detail}"
            )
            continue
        preview = value if len(value) <= 60 else value[:57] + "..."
        log(f"PATCHed /{resource}/{loc_id} ({locale}) {field}={preview!r}")
        written += 1
    return written


def _apply_locale(
    token: str,
    locale: str,
    loc_state: dict[str, Any],
    ai_locale: dict[str, Any],
    empty_list: list[str],
) -> int:
    """PATCH the writes for one locale. Returns count of fields written."""
    app_writes = _build_writes(ai_locale, empty_list, APP_LEVEL_FIELDS)
    ver_writes = _build_writes(ai_locale, empty_list, VERSION_LEVEL_FIELDS)
    _describe_with_footer(locale, loc_state, ver_writes)
    return (
        _patch_group(token, APP_INFO_RESOURCE, locale,
                     loc_state.get("app_info_localization_id"), app_writes)
        + _patch_group(token, VERSION_RESOURCE, locale,
                       loc_state.get("version_localization_id"), ver_writes)
    )


def apply(
    state: dict[str, Any],
    ai_json: dict[str, Any],
    token: str,
    fields_filter: set[str] | None = None,
) -> tuple[int, int]:
    """Drive `_apply_locale` across every locale in `state['empty_fields']`.

    When ``fields_filter`` is provided, only fields in the set are considered
    -- the detector's empty_list is intersected with the filter before gating,
    preserving the double-gate invariant (detector must still have flagged the
    field as empty). Used by the two-phase orchestrator to PATCH
    ``description`` first (phase 1) then the remaining fields (phase 2).

    Returns (total_fields_written, locales_touched).
    """
    empty_fields = state.get("empty_fields") or {}
    localizations = state.get("localizations") or {}
    ai_locs = (ai_json or {}).get("localizations") or {}

    total_written = 0
    locales_touched = 0
    for locale, empty_list in empty_fields.items():
        if not empty_list:
            continue
        effective = (
            [f for f in empty_list if f in fields_filter]
            if fields_filter is not None
            else list(empty_list)
        )
        if not effective:
            continue
        loc_state = localizations.get(locale) or {}
        ai_locale = ai_locs.get(locale) or {}
        if not ai_locale:
            continue
        written = _apply_locale(token, locale, loc_state, ai_locale, effective)
        if written:
            total_written += written
            locales_touched += 1
    return total_written, locales_touched


def _parse_fields_filter(raw: str | None) -> set[str] | None:
    """Parse the ``--fields-filter`` CLI arg into a set or None."""
    if not raw:
        return None
    parsed = {f.strip() for f in raw.split(",") if f.strip()}
    return parsed or None


def main() -> int:
    parser = argparse.ArgumentParser(description="Apply AI-generated ASC metadata")
    parser.add_argument("--state", required=True, help="detector state JSON")
    parser.add_argument("--response", required=True, help="AI response JSON")
    parser.add_argument(
        "--fields-filter",
        default=None,
        help="comma-separated subset of fields to PATCH (intersected with empty_fields)",
    )
    args = parser.parse_args()

    fields_filter = _parse_fields_filter(args.fields_filter)
    if fields_filter is None:
        log("field-filter: none (all empty fields)")
    else:
        log(f"field-filter active: {sorted(fields_filter)}")

    state = _load_json(args.state)
    ai_json = _load_json(args.response)
    if state is None or ai_json is None:
        return 0

    # Normalize any unescaped control chars / multi-line runs the AI may
    # have emitted -- keeps field values within Apple's char limits and
    # prevents literal newlines leaking into the store listing.
    ai_json = _normalize_ai_response(ai_json)

    if not state.get("empty_fields"):
        log("metadata fully populated; AI applier no-op")
        return 0
    if not (ai_json.get("localizations") or {}):
        warn("AI response contains no localizations; nothing to apply")
        return 0

    try:
        token = make_jwt(
            require_env("ASC_KEY_ID"),
            require_env("ASC_ISSUER_ID"),
            require_env("ASC_KEY_PATH"),
        )
    except SystemExit:
        raise
    except Exception as exc:
        warn(f"ASC JWT generation failed: {exc!r}")
        return 0

    written, locales = apply(state, ai_json, token, fields_filter=fields_filter)
    log(f"applied {written} writes across {locales} locales")
    return 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except SystemExit:
        raise
    except Exception as exc:  # fail-open parity with set_app_store_whats_new
        warn(f"asc_metadata_applier failed (non-fatal): {exc!r}")
        sys.exit(0)