license_utils.py

import logging
import re
from dataclasses import dataclass
from difflib import SequenceMatcher
from sqlalchemy.orm import Session
from sqlalchemy import select, func
from typing import List, Tuple, Optional

from shared.common.db_utils import normalize_url, normalize_url_str
from shared.database_gen.sqlacodegen_models import License, FeedLicenseChange, Feed


@dataclass
class MatchingLicense:
    """Response structure for license URL resolution.

    Represents a matched license result from the resolution process, containing
    identification, matching metadata, and confidence scoring.

    Attributes:
        license_id: Unique identifier for the license (typically SPDX ID)
        license_url: Original license URL provided for resolution
        normalized_url: URL after normalization (lowercased, trimmed, protocol removed)
        match_type: Type of match performed. One of:
            - 'exact': Direct match found in database
            - 'heuristic': Matched via pattern-based rules (CC resolver, common patterns)
            - 'fuzzy': Similarity-based match against same-host licenses
            - 'none': No match found
        confidence: Match confidence score (0.0-1.0)
            - 1.0: Exact match
            - 0.99: Creative Commons resolved
            - 0.95: Pattern heuristic match
            - 0.0-1.0: Fuzzy match score based on string similarity
        spdx_id: SPDX License Identifier if matched (e.g., 'CC-BY-4.0', 'MIT')
        matched_name: Human-readable name of the matched license
        matched_catalog_url: Canonical URL from the license catalog/database
        matched_source: Source of the match. One of:
            - 'db.license': Exact match from database
            - 'cc-resolver': Creative Commons license resolver
            - 'pattern-heuristics': Generic pattern matching
        notes: Additional context about the match (e.g., version normalization, locale detection)
        regional_id: Regional/jurisdictional variant identifier for ported licenses
            (e.g., 'CC-BY-2.1-jp' for Japan-ported Creative Commons)
    """

    license_id: str
    license_url: str
    normalized_url: str
    match_type: str
    confidence: float
    spdx_id: str | None = None
    matched_name: str | None = None
    matched_catalog_url: str | None = None
    matched_source: str | None = None
    notes: str | None = None
    regional_id: str | None = None


# The COMMON_PATTERNS list contains tuples of (regex pattern, SPDX ID).
# It is used for heuristic matching of license URLs.
COMMON_PATTERNS = [
    (re.compile(r"opendatacommons\.org/licenses/odbl/1\.0/?", re.I), "ODbL-1.0"),
    (re.compile(r"opendatacommons\.org/licenses/by/1\.0/?", re.I), "ODC-By-1.0"),
    (re.compile(r"opendatacommons\.org/licenses/pddl/1\.0/?", re.I), "PDDL-1.0"),
    (re.compile(r"opensource\.org/licenses/Apache-2\.0/?", re.I), "Apache-2.0"),
    (re.compile(r"opensource\.org/licenses/MIT/?", re.I), "MIT"),
    (re.compile(r"choosealicense\.com/licenses/mit/?", re.I), "MIT"),
    (re.compile(r"choosealicense\.com/licenses/apache-2\.0/?", re.I), "Apache-2.0"),
]


def extract_host(url: str) -> str:
    """Extract host only from normalized URL."""
    # if the url has protocol like http://, normalize_url_str should have removed it
    normalized_url = normalize_url_str(url)
    return normalized_url.split("/", 1)[0] if normalized_url else ""


def resolve_commons_creative_license(url: str) -> Tuple[Optional[str], Optional[str], Optional[str]]:
    """
    Resolve a Creative Commons license URL to an SPDX ID and an explanatory note.

    Returns:
        (spdx_id, note)
          - spdx_id: SPDX identifier string if resolved, else None
          - note: additional context (e.g., locale port detected, version normalized), else None
          - regional_id: locale/ported variant if present (e.g., 'CC-BY-2.1-jp'), else None

    Behavior & Rationale:
    ---------------------
    1) Normalizes common CC URL variants
       - Creative Commons pages often add suffixes like '/deed', '/deed.<lang>', '/legalcode', '/legalcode.<lang>'.
         These suffixes are *presentation pages*, not distinct licenses. We strip them before matching.

    2) Handles CC0 explicitly
       - CC0 is under 'publicdomain/zero/1.0/'. The SPDX ID is 'CC0-1.0'.

    3) Parses CC license family, version, and optional locale/port
       - Pattern matched: 'creativecommons.org/licenses/<code>/<version>[/<locale>]'
         Examples:
           https://creativecommons.org/licenses/by/4.0/
           https://creativecommons.org/licenses/by/2.5/de/
           https://creativecommons.org/licenses/by-nc-sa/3.0/jp/deed.ja

       - <code> is one of: by, by-sa, by-nd, by-nc, by-nc-sa, by-nc-nd
       - <version> is a dotted number like 1.0, 2.0, 2.1, 2.5, 3.0, 4.0
       - <locale> historically denotes a jurisdiction "port" (e.g., 'jp', 'fr', 'de').

    4) Locale ports (jurisdiction-specific variants) are *not* in the SPDX License List
       - Creative Commons no longer recommends using ported licenses.
       - SPDX lists canonical (global) CC licenses and not the ported variants.
       - If a locale segment is present, we keep the canonical family/version and add a note explaining that the port
         was detected and ignored.

    5) Version normalization
       - SPDX includes certain CC versions (1.0, 2.0, 2.5, 3.0, 4.0) but *not* 2.1.
       - Some historical ported pages use "2.1" (e.g., '.../by/2.1/jp/'). Map these to the closest SPDX-supported
         equivalent: **2.1 → 2.0** for all CC BY* families.
       - If an unexpected version appears, we attempt a conservative normalization to the closest known version,
         preferring the nearest *lower or equal* recognized version (4.0, 3.0, 2.5, 2.0, 1.0). A note explains this.

    Examples:
      - https://creativecommons.org/licenses/by/2.1/jp/  →
        ("CC-BY-2.0", "Detected locale 'jp' and normalized 2.1 → 2.0.")
      - https://creativecommons.org/licenses/by-sa/2.5/de →
        ("CC-BY-SA-2.5", "Detected locale 'de' (ported license ignored).")
      - https://creativecommons.org/licenses/by/4.0/      → ("CC-BY-4.0", None)
      - https://creativecommons.org/publicdomain/zero/1.0/legalcode.en → ("CC0-1.0", None)
    """

    # Use your existing string-normalization utility (assumed to:
    # - lowercase host, strip fragments/query/whitespace, collapse slashes, etc.)
    n = normalize_url_str(url)

    # Remove presentation-only CC suffixes like '/legalcode', '/legalcode.xx', '/deed', '/deed.xx'
    n = re.sub(r"/legalcode(\.[a-zA-Z\-]+)?$", "", n, flags=re.I)
    n = re.sub(r"/deed(\.[a-zA-Z\-]+)?$", "", n, flags=re.I)

    # --- CC0 special case -----------------------------------------------------
    if re.search(r"creativecommons\.org/publicdomain/zero/1\.0/?$", n, re.I):
        return "CC0-1.0", None, None

    # --- General CC licenses --------------------------------------------------
    # Capture family code, version, and optional locale (jurisdiction port).
    # Locale historically tends to be 2 letters, but allow 2–5 just in case (e.g., 'pt-br').
    m = re.search(
        r"creativecommons\.org/licenses/([a-z\-]+)/([\d\.]+)(?:/([a-z\-]{2,5}))?/?$",
        n,
        re.I,
    )
    if not m:
        return None, None, None

    code = m.group(1).lower()  # e.g., 'by', 'by-sa', 'by-nc-nd'
    ver_in = m.group(2)  # e.g., '2.5'
    locale = m.group(3)  # e.g., 'jp', 'fr', 'de', or None

    # Map CC family code to SPDX base
    family_map = {
        "by": "CC-BY",
        "by-sa": "CC-BY-SA",
        "by-nd": "CC-BY-ND",
        "by-nc": "CC-BY-NC",
        "by-nc-sa": "CC-BY-NC-SA",
        "by-nc-nd": "CC-BY-NC-ND",
    }
    base = family_map.get(code)
    if not base:
        return None, None, None

    note_parts = []

    # If a locale/jurisdiction port is present, record a note and ignore it for SPDX ID construction.
    if locale:
        note_parts.append(
            f"Detected locale/jurisdiction port '{locale}'. SPDX does not list ported CC licenses; using canonical ID."
        )

    # Normalize version to nearest SPDX-supported version.
    # Direct map for commonly-seen CC versions and the special 2.1 → 2.0 case.
    direct_version_map = {
        "1.0": "1.0",
        "2.0": "2.0",
        "2.1": "2.0",  # CC 2.1 ports are mapped to the closest SPDX-supported version (2.0)
        "2.5": "2.5",
        "3.0": "3.0",
        "4.0": "4.0",
    }

    if ver_in in direct_version_map:
        ver_out = direct_version_map[ver_in]
        if ver_out != ver_in:
            note_parts.append(f"Normalized version {ver_in} → {ver_out} to match SPDX-supported versions.")
    else:
        # Fallback: choose the closest *lower or equal* known version.
        # (Most unknowns should still land on an SPDX-supported canonical version.)
        known = ["4.0", "3.0", "2.5", "2.0", "1.0"]
        ver_out = None
        try:
            vin = float(ver_in)
            # pick the highest known <= vin, else default to the lowest (1.0)
            candidates = [kv for kv in known if vin >= float(kv)]
            ver_out = candidates[0] if candidates else known[-1]
        except ValueError:
            # Non-numeric (unexpected) — choose the most modern canonical as a pragmatic default
            ver_out = "4.0"

        note_parts.append(f"Unrecognized CC version '{ver_in}'. Chose closest canonical version '{ver_out}' for SPDX.")

    spdx_id = f"{base}-{ver_out}"
    regional_id = f"{base}-{ver_in}-{locale.lower()}" if locale else None
    note = " ".join(note_parts) if note_parts else None
    return spdx_id, note, regional_id


def heuristic_spdx(url: str) -> str | None:
    """Heuristic SPDX resolver based on common URL patterns."""
    for rx, spdx in COMMON_PATTERNS:
        if rx.search(url) or rx.search(normalize_url_str(url)):
            return spdx
    return None


def fuzzy_ratio(a: str, b: str) -> float:
    """Compute fuzzy similarity ratio between two strings."""
    return SequenceMatcher(None, a, b).ratio()


def resolve_fuzzy_match(
    url_str: str,
    url_host: str,
    url_normalized: str,
    fuzzy_threshold: float,
    db_session: Session | None = None,
    max_candidates: int | None = 5,
) -> List[MatchingLicense]:
    """Fuzzy match license URL against same-host candidates in DB.

    Returns a sorted list of candidates (best first) whose similarity is >= fuzzy_threshold.
    """
    if not db_session or not url_host:
        return []

    # Pull candidates from DB and filter by host in Python, based on License.url
    db_licenses: list[License] = list(db_session.scalars(select(License)))
    same_host: list[License] = []
    for lic in db_licenses:
        if not getattr(lic, "url", None):
            continue
        if extract_host(normalize_url_str(lic.url)) == url_host:
            same_host.append(lic)

    scored: list[tuple[float, License]] = []
    for lic in same_host:
        lic_norm = normalize_url_str(lic.url)
        score = fuzzy_ratio(url_normalized, lic_norm)
        if score >= fuzzy_threshold:
            scored.append((float(score), lic))

    # Sort by descending score and optionally limit
    scored.sort(key=lambda x: x[0], reverse=True)
    if max_candidates is not None:
        scored = scored[:max_candidates]

    results: List[MatchingLicense] = []
    for score, lic in scored:
        results.append(
            MatchingLicense(
                license_id=lic.id,
                license_url=url_str,
                normalized_url=url_normalized,
                spdx_id=lic.id,
                match_type="fuzzy",
                confidence=round(score, 3),
                matched_name=lic.name,
                matched_catalog_url=lic.url,
                matched_source="db.license",
            )
        )
    return results


def find_exact_match_license_url(url_normalized: str, db_session: Session | None) -> License | None:
    """Find exact match of normalized license URL in DB (License.url)."""
    if not db_session:
        return None
    # Compare normalized strings using SQL functions on License.url
    return (
        db_session.query(License)
        .filter(normalize_url_str(url_normalized) == func.lower(func.trim(normalize_url(License.url))))
        .first()
    )


def extract_spdx_id_from_url(url_normalized: str) -> Optional[str]:
    """Extract an SPDX license ID from an SPDX-style URL if present.

    Recognizes URLs of the form used on spdx.org, for example::

        https://spdx.org/licenses/ODbL-1.0.html
        http://spdx.org/licenses/MIT

    The function is conservative and only returns an SPDX ID when it finds a
    path segment under ``/licenses/`` that looks like an SPDX identifier. Any
    optional ``.html`` suffix is stripped.
    """
    # Match host 'spdx.org' and capture the token after '/licenses/' up to
    # an optional '.html' suffix and optional trailing slash.
    match = re.search(r"spdx\.org/licenses/([^/?#]+?)(?:\.html)?/?$", url_normalized, re.I)
    if not match:
        return None

    spdx_id = match.group(1)
    # Basic sanity check: SPDX IDs are typically alnum plus '-', '.' (e.g. 'CC-BY-4.0')
    if not re.fullmatch(r"[A-Za-z0-9.+-]+", spdx_id):
        return None

    return spdx_id


def resolve_license(
    license_url: str,
    allow_fuzzy: bool = True,
    fuzzy_threshold: float = 0.94,
    db_session: Session | None = None,
) -> List[MatchingLicense]:
    """Resolve a license URL to one or more SPDX candidates using multiple strategies.

    Strategies (in order of precedence):
      1) Exact match in DB (``db.license``)                 -> return [exact]
      2) Creative Commons resolver (``cc-resolver``)       -> return [cc]
      3) SPDX catalog URL resolver (``spdx.org/licenses``) -> return [spdx]
      4) Generic heuristics (pattern-based)                -> return [heuristic]
      5) Fuzzy (same-host candidates)                      -> return [fuzzy...]
      6) No match                                          -> return []

    Args:
        license_url (str): The license URL to resolve.
        allow_fuzzy (bool): Whether to allow fuzzy matching.
        fuzzy_threshold (float): Minimum similarity ratio for fuzzy match.
        db_session (Session | None): SQLAlchemy DB session. Required for DB-based strategies.

    Returns:
        List[MatchingLicense]: Ordered list of resolution results. Empty if no match.
    """
    url_str = str(license_url)
    url_normalized = normalize_url_str(url_str)
    url_host = extract_host(url_normalized)

    # 1) Exact hit in DB (compare normalized strings of known licenses)
    exact_match: License | None = find_exact_match_license_url(url_normalized, db_session) if db_session else None
    if exact_match:
        return [
            MatchingLicense(
                license_id=exact_match.id,
                license_url=url_str,
                normalized_url=url_normalized,
                spdx_id=exact_match.id,
                match_type="exact",
                confidence=1.0,
                matched_name=exact_match.name,
                matched_catalog_url=exact_match.url,
                matched_source="db.license",
            )
        ]

    # 2) Creative Commons resolver
    common_creative_match, notes, regional_id = resolve_commons_creative_license(url_str)
    if common_creative_match:
        cc_license: License | None = db_session.query(License).filter(License.id == common_creative_match).one_or_none()
        if not cc_license:
            logging.warning("CC license SPDX ID %s not found in DB", common_creative_match)
            return []
        return [
            MatchingLicense(
                license_id=cc_license.id,
                license_url=url_str,
                normalized_url=url_normalized,
                spdx_id=common_creative_match,
                match_type="heuristic",
                confidence=0.99,
                # Fill in matched_name with SPDX ID for lack of better info
                matched_name=common_creative_match,
                matched_catalog_url=None,
                matched_source="cc-resolver",
                notes=notes,
                regional_id=regional_id,
            )
        ]

    # 3) SPDX catalog URL (spdx.org/licenses/<ID>[.html])
    spdx_id = extract_spdx_id_from_url(url_normalized)
    if spdx_id:
        # Try to enrich from DB if a matching License row exists
        db_lic: License | None = (
            db_session.query(License).filter(func.lower(License.id) == func.lower(spdx_id)).one_or_none()
        )
        if db_lic is not None:
            return [
                MatchingLicense(
                    license_id=db_lic.id,
                    license_url=url_str,
                    normalized_url=url_normalized,
                    spdx_id=spdx_id,
                    match_type="heuristic",
                    confidence=0.98,
                    matched_name=db_lic.name,
                    matched_catalog_url=db_lic.url,
                    matched_source="spdx-resolver",
                )
            ]
        else:
            logging.warning("SPDX ID %s resolved from URL but not found in DB", spdx_id)

    # 4) Generic heuristics
    heuristic_match = heuristic_spdx(url_str)
    if heuristic_match:
        if db_session is not None:
            # Check if the license found is actually in the DB
            db_lic = (
                db_session.query(License).filter(func.lower(License.id) == func.lower(heuristic_match)).one_or_none()
            )
            if db_lic is None:
                logging.warning("Heuristic SPDX ID %s not found in DB, skipping assignment", heuristic_match)
                heuristic_match = None
        if heuristic_match:
            return [
                MatchingLicense(
                    license_id=heuristic_match,
                    license_url=url_str,
                    normalized_url=url_normalized,
                    spdx_id=heuristic_match,
                    match_type="heuristic",
                    confidence=0.95,
                    matched_name=heuristic_match,
                    matched_source="pattern-heuristics",
                )
            ]

    # 5) Fuzzy (same host candidates only)
    if allow_fuzzy and url_host and db_session is not None:
        fuzzy_results = resolve_fuzzy_match(
            url_str=url_str,
            url_host=url_host,
            url_normalized=url_normalized,
            fuzzy_threshold=fuzzy_threshold,
            db_session=db_session,
        )
        if fuzzy_results:
            return fuzzy_results

    # 6) No match
    return []


# Confidence threshold above which an auto-assigned license is considered verified
# without requiring human review. Covers exact, CC resolver, SPDX, and pattern heuristic matches.
_AUTO_VERIFY_THRESHOLD = 0.95


def assign_license_by_url(
    feed,
    db_session: Session,
    *,
    only_if_single: bool = True,
) -> Optional[MatchingLicense]:
    """Resolve feed.license_url and auto-assign a license if exactly one match is found.

    Behavior:
    - 0 matches: logs info, returns None (no change).
    - >1 matches: logs a warning and returns None when only_if_single=True;
      the feed retains its current license_id for manual review.
    - 1 match: assigns feed.license_id / feed.license_notes and appends a
      FeedLicenseChange audit row. verified is set based on confidence:
        - True  if match_type == 'exact' or confidence >= _AUTO_VERIFY_THRESHOLD
          (covers exact DB matches, CC resolver, SPDX, pattern heuristics)
        - False if match_type == 'fuzzy' (needs human confirmation)

    Args:
        feed: Any Feed ORM instance (Gtfsfeed, Gtfsrealtimefeed, Gbfsfeed).
        db_session: Active SQLAlchemy session; required for DB-backed resolution.
        only_if_single: When True (default), skip assignment if multiple candidates
            are returned, requiring a human to choose.

    Returns:
        The assigned MatchingLicense, or None if no assignment was made.
    """
    if not feed.license_url:
        return None

    matches = resolve_license(feed.license_url, db_session=db_session)

    if not matches:
        logging.info(
            "No license match found for feed %s (url: %s)",
            feed.stable_id,
            feed.license_url,
        )
        return None

    if only_if_single and len(matches) > 1:
        logging.warning(
            "Skipping auto-assignment for feed %s: %d license candidates found — manual review required",
            feed.stable_id,
            len(matches),
        )
        return None

    best = matches[0]

    if best.license_id == feed.license_id:
        logging.info("Feed %s license unchanged: %s", feed.stable_id, best.license_id)
        return best

    is_verified = best.match_type == "exact" or best.confidence >= _AUTO_VERIFY_THRESHOLD

    logging.info(
        "Assigning license %s to feed %s (match_type=%s, confidence=%.2f, verified=%s)",
        best.license_id,
        feed.stable_id,
        best.match_type,
        best.confidence,
        is_verified,
    )

    feed.license_id = best.license_id
    feed.license_notes = best.notes
    feed.feed_license_changes.append(
        FeedLicenseChange(
            feed_id=feed.id,
            changed_at=None,  # set by DB default
            feed_license_url=feed.license_url,
            matched_license_id=best.license_id,
            confidence=best.confidence,
            match_type=best.match_type,
            matched_name=best.matched_name,
            matched_catalog_url=best.matched_catalog_url,
            matched_source=best.matched_source,
            notes=best.notes,
            regional_id=best.regional_id,
            verified=is_verified,
        )
    )

    return best


@dataclass
class PropagateLicenseAffectedFeedResult:
    """Describes a single feed affected by a license propagation."""

    feed_id: str
    previous_license_id: Optional[str]
    data_type: Optional[str]


@dataclass
class PropagateLicenseResult:
    """Result of a license propagation operation.

    Attributes:
        license_id: The license ID that was propagated.
        license_url: The original license URL provided for matching.
        normalized_license_url: Normalized form of the license URL used for matching.
        dry_run: Whether this was a dry-run (no changes persisted).
        override: Whether feeds with an existing license_id were also updated.
        total_feeds_with_same_url: Total feeds sharing the same normalized license URL.
        affected_feeds_count: Number of feeds that were (or would be) updated.
        affected_feeds: List of affected feed descriptors.
    """

    license_id: str
    license_url: str
    normalized_license_url: str
    dry_run: bool
    override: bool
    total_feeds_with_same_url: int
    affected_feeds_count: int
    affected_feeds: List[PropagateLicenseAffectedFeedResult]


def propagate_license_by_url(
    license_id: str,
    license_url: str,
    db_session: Session,
    *,
    dry_run: bool = True,
    override: bool = False,
) -> PropagateLicenseResult:
    """Propagate a license ID to all feeds sharing the same normalized license URL.

    Finds all published (non-unpublished) feeds whose license_url normalizes to the
    same value as ``license_url``, then optionally updates their ``license_id`` and
    creates ``FeedLicenseChange`` audit records.

    Args:
        license_id: The license ID to propagate. Must exist in the ``license`` table.
        license_url: The reference URL whose normalized form is used for matching.
        db_session: Active SQLAlchemy session.
        dry_run: When True (default), compute results without persisting changes.
        override: When False (default), only update feeds where ``license_id IS NULL``.
                  When True, also update feeds that already have a different ``license_id``.

    Returns:
        A ``PropagateLicenseResult`` describing the outcome.

    Raises:
        ValueError: If ``license_id`` does not exist in the database.
    """
    existing_license = db_session.get(License, license_id)
    if existing_license is None:
        raise ValueError(f"License '{license_id}' not found in the database.")

    normalized_url = normalize_url_str(license_url)

    # Find all feeds with the same normalized license URL.
    # Use the same SQL normalization pattern as get_feed_query_by_normalized_url.
    candidate_query = db_session.query(Feed).filter(
        Feed.license_url.isnot(None),
        Feed.operational_status != "unpublished",
        normalized_url == func.lower(func.trim(normalize_url(Feed.license_url))),
    )
    all_candidates = candidate_query.all()
    total_feeds_with_same_url = len(all_candidates)

    if override:
        feeds_to_update = [f for f in all_candidates if f.license_id != license_id]
    else:
        feeds_to_update = [f for f in all_candidates if f.license_id is None]

    affected: List[PropagateLicenseAffectedFeedResult] = []
    for feed in feeds_to_update:
        affected.append(
            PropagateLicenseAffectedFeedResult(
                feed_id=feed.stable_id,
                previous_license_id=feed.license_id,
                data_type=feed.data_type,
            )
        )
        if not dry_run:
            feed.license_id = license_id
            db_session.add(
                FeedLicenseChange(
                    feed_id=feed.id,
                    feed_license_url=feed.license_url,
                    matched_license_id=license_id,
                    confidence=1.0,
                    match_type="propagated",
                    matched_source="propagate_match",
                    verified=True,
                )
            )

    logging.info(
        "propagate_license_by_url: license_id=%s url=%s dry_run=%s override=%s " "total_with_url=%d affected=%d",
        license_id,
        license_url,
        dry_run,
        override,
        total_feeds_with_same_url,
        len(affected),
    )

    return PropagateLicenseResult(
        license_id=license_id,
        license_url=license_url,
        normalized_license_url=normalized_url,
        dry_run=dry_run,
        override=override,
        total_feeds_with_same_url=total_feeds_with_same_url,
        affected_feeds_count=len(affected),
        affected_feeds=affected,
    )