Create commercial_stats.py

commercial_stats.py

@@ -0,0 +1,171 @@
+# commercial_stats.py
+import json
+import os
+import re
+import subprocess
+import tempfile
+from typing import Any, Dict, List, Tuple, Optional
+
+import requests
+from fastapi import APIRouter
+from fastapi.responses import JSONResponse
+
+from data import flatten_models, license_label_and_url
+
+router = APIRouter()
+
+SCANCODE_CMD = os.environ.get("SCANCODE_CMD", "scancode")  # allow override if needed
+HTTP_TIMEOUT = 25
+
+# Very light categorization strictly from SPDX/CC ids detected by ScanCode.
+# No guesswork/fallback detection—just label grouping for the stats panel.
+def _bucket_spdx(spdx_ids: List[str]) -> Dict[str, bool]:
+    s = " ".join(spdx_ids).upper()
+    is_noncommercial = ("-NC" in s) or ("NON-COMMERCIAL" in s)
+    is_no_derivatives = ("-ND" in s) or ("NO-DERIVATIVE" in s)
+    is_copyleft = bool(re.search(r"\b(AGPL|GPL|LGPL|MPL)\b", s))
+    # “Commercial-friendly” = not NC, not ND. Copyleft may still allow commercial use;
+    # we keep it separate so the UI can show both signals.
+    is_commercial_friendly = not is_noncommercial and not is_no_derivatives
+    return {
+        "commercial_friendly": is_commercial_friendly,
+        "noncommercial": is_noncommercial,
+        "no_derivatives": is_no_derivatives,
+        "copyleft": is_copyleft,
+    }
+
+def _run_scancode_on_text(text: str) -> Tuple[List[str], List[str]]:
+    """
+    Run ScanCode CLI on a temp file containing `text`.
+    Returns (spdx_ids, keys) where:
+      - spdx_ids: list of detected SPDX license identifiers
+      - keys:     ScanCode 'key' values (their internal identifiers)
+    """
+    with tempfile.TemporaryDirectory() as td:
+        path = os.path.join(td, "LICENSE.txt")
+        out_json = os.path.join(td, "out.json")
+        with open(path, "w", encoding="utf-8") as f:
+            f.write(text)
+        # CLI: detect licenses only, output JSON
+        cmd = [
+            SCANCODE_CMD,
+            "--license",
+            "--json-pp", out_json,
+            path,
+        ]
+        try:
+            proc = subprocess.run(
+                cmd, check=True, capture_output=True, text=True, timeout=120
+            )
+        except subprocess.CalledProcessError as e:
+            raise RuntimeError(f"ScanCode failed: {e.stderr or e.stdout}") from e
+        except Exception as e:
+            raise
+
+        with open(out_json, "r", encoding="utf-8") as f:
+            data = json.load(f)
+
+        # Extract SPDX ids and ScanCode keys from results
+        spdx_ids: List[str] = []
+        keys: List[str] = []
+        for res in data.get("files", []):
+            for lic in res.get("licenses", []):
+                spdx = lic.get("spdx_license_key") or ""
+                key = lic.get("key") or ""
+                if spdx:
+                    spdx_ids.append(spdx)
+                if key:
+                    keys.append(key)
+        return (sorted(set(spdx_ids)), sorted(set(keys)))
+
+def _fetch_license_text(url: str) -> Optional[str]:
+    try:
+        r = requests.get(url, timeout=HTTP_TIMEOUT)
+        if r.status_code == 200 and r.text:
+            return r.text
+    except Exception:
+        pass
+    return None
+
+@router.get("/commercial/stats")
+def commercial_stats():
+    models = flatten_models()
+    total = len(models)
+    analyzed = 0
+    unknown = 0
+
+    # Aggregate by buckets and by SPDX id for transparency
+    buckets = {
+        "commercial_friendly": 0,
+        "noncommercial": 0,
+        "no_derivatives": 0,
+        "copyleft": 0,
+    }
+    by_spdx: Dict[str, int] = {}
+
+    per_model: List[Dict[str, Any]] = []
+
+    for m in models:
+        label, url = license_label_and_url(m)
+        if not url:
+            unknown += 1
+            per_model.append({
+                "id": f"{m.get('owner') or m.get('username') or ''}/{m.get('name') or ''}".strip("/"),
+                "license_label": label,
+                "spdx": [],
+                "buckets": {},
+                "source": None
+            })
+            continue
+
+        text = _fetch_license_text(url)
+        if not text:
+            unknown += 1
+            per_model.append({
+                "id": f"{m.get('owner') or m.get('username') or ''}/{m.get('name') or ''}".strip("/"),
+                "license_label": label,
+                "spdx": [],
+                "buckets": {},
+                "source": url
+            })
+            continue
+
+        try:
+            spdx_ids, _keys = _run_scancode_on_text(text)
+        except Exception:
+            # No fallback detection—just mark unknown
+            unknown += 1
+            per_model.append({
+                "id": f"{m.get('owner') or m.get('username') or ''}/{m.get('name') or ''}".strip("/"),
+                "license_label": label,
+                "spdx": [],
+                "buckets": {},
+                "source": url
+            })
+            continue
+
+        analyzed += 1
+        for sid in spdx_ids:
+            by_spdx[sid] = by_spdx.get(sid, 0) + 1
+
+        b = _bucket_spdx(spdx_ids)
+        for k, v in b.items():
+            if v:
+                buckets[k] += 1
+
+        per_model.append({
+            "id": f"{m.get('owner') or m.get('username') or ''}/{m.get('name') or ''}".strip("/"),
+            "license_label": label,
+            "spdx": spdx_ids,
+            "buckets": b,
+            "source": url
+        })
+
+    return JSONResponse({
+        "total_models": total,
+        "analyzed": analyzed,
+        "unknown": unknown,
+        "buckets": buckets,
+        "by_spdx": dict(sorted(by_spdx.items(), key=lambda kv: kv[1], reverse=True)),
+        "per_model": per_model[:200]  # cap for payload size
+    })