From ecb866f111f087f8dec879b3406e4c6d23c7b751 Mon Sep 17 00:00:00 2001 From: Seungpyo1007 Date: Fri, 4 Sep 2026 15:23:54 +0900 Subject: [PATCH] feat(app): classify ES vs QS vs retail offline --- README.md | 12 ++ app/ingest/__init__.py | 5 + app/ingest/__main__.py | 10 ++ app/ingest/classify.py | 274 +++++++++++++++++++++++++++++++++++++++++ tests/test_classify.py | 128 +++++++++++++++++++ 5 files changed, 429 insertions(+) create mode 100644 app/ingest/__init__.py create mode 100644 app/ingest/__main__.py create mode 100644 app/ingest/classify.py create mode 100644 tests/test_classify.py diff --git a/README.md b/README.md index 46241dc..6c19697 100644 --- a/README.md +++ b/README.md @@ -36,6 +36,18 @@ python -m pytest -q The validator uses the Python standard library. QS, retail, and production `sample_class` values fail the build. +## Collection + +Intake is **classify-then-write**. A candidate is `es`, `qs`, `retail`, or `unknown`. Only `es` may become a JSON file. QS and retail stay out; `unknown` is for human review. + +```bash +python -m app.ingest path/to/candidate.json +``` + +No network. Identifier rules: Intel S-spec → retail, Q-spec + early stepping / low clocks / “Intel Confidential” → es, Q-spec labeled QS or mature stepping with retail clocks → qs. AMD `100-00000…` / Eng Sample → es, historical `Z…` OPN → qs. + +Crawlers come later. This classifier is the gate. + ## Branching (git-flow) | Branch | Role | diff --git a/app/ingest/__init__.py b/app/ingest/__init__.py new file mode 100644 index 0000000..97d8b5a --- /dev/null +++ b/app/ingest/__init__.py @@ -0,0 +1,5 @@ +"""ES intake: classify public identifiers before any JSON is written.""" + +from app.ingest.classify import Classification, classify + +__all__ = ["Classification", "classify"] diff --git a/app/ingest/__main__.py b/app/ingest/__main__.py new file mode 100644 index 0000000..99dfb50 --- /dev/null +++ b/app/ingest/__main__.py @@ -0,0 +1,10 @@ +"""``python -m app.ingest`` classifies JSON on stdin or a file path.""" + +from __future__ import annotations + +import sys + +from app.ingest.classify import main + +if __name__ == "__main__": + sys.exit(main()) diff --git a/app/ingest/classify.py b/app/ingest/classify.py new file mode 100644 index 0000000..b911824 --- /dev/null +++ b/app/ingest/classify.py @@ -0,0 +1,274 @@ +"""Offline ES / QS / retail classifier. + +No network. A candidate must be ``es`` before the ingest pipeline may write JSON. +QS and retail belong in TechAPI or nowhere; ``unknown`` goes to a review issue. +""" + +from __future__ import annotations + +import argparse +import json +import re +import sys +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Any + +CLASS_ES = "es" +CLASS_QS = "qs" +CLASS_RETAIL = "retail" +CLASS_UNKNOWN = "unknown" + +# Intel production S-spec: 5 letters starting with S (e.g. SRKNY). +_SSPEC_RE = re.compile(r"^S[A-Z0-9]{4}$") +# Intel sample QDF / Q-spec: 4–6 letters starting with Q (e.g. QXLB, QDF4). +_QSPEC_RE = re.compile(r"^Q[A-Z0-9]{3,5}$") +# Zen 4+ engineering OPN, e.g. 100-000000665-21_N +_AMD_MODERN_OPN_RE = re.compile( + r"^100-0+\d+(?:-\d+)?(?:_[A-Z0-9]+)?$", re.IGNORECASE +) + +_ES_STEPPINGS = {"A0", "B0", "G0"} +_QS_STEPPINGS = {"C0", "H0"} + +_QS_PHRASES = ( + "qualification sample", + "qualification-sample", + " qs ", + "(qs)", + "[qs]", +) +_ES_PHRASES = ( + "engineering sample", + "eng sample", + "eng. sample", + "intel confidential", +) +_RETAIL_PHRASES = ("retail", "production sku", "production part") + + +@dataclass(frozen=True) +class Classification: + """Result of classifying one candidate. ``sample_class`` is never empty.""" + + sample_class: str + reasons: tuple[str, ...] + sample_revision: str | None = None + + def to_dict(self) -> dict[str, Any]: + return asdict(self) + + +def _norm(value: object) -> str: + if not isinstance(value, str): + return "" + return value.strip() + + +def _upper(value: object) -> str: + return _norm(value).upper() + + +def _blob(candidate: dict[str, Any]) -> str: + parts: list[str] = [] + for key in ("name", "slug", "notes", "sample_class"): + parts.append(_norm(candidate.get(key))) + markings = candidate.get("markings") + if isinstance(markings, list): + parts.extend(_norm(m) for m in markings) + labels = candidate.get("source_labels") + if isinstance(labels, list): + parts.extend(_norm(label) for label in labels) + return " ".join(parts).lower() + + +def _has_phrase(blob: str, phrases: tuple[str, ...]) -> str | None: + padded = f" {blob} " + for phrase in phrases: + if phrase in padded or phrase.strip() in blob: + return phrase.strip() + return None + + +def _intel_qspec(candidate: dict[str, Any]) -> str: + qspec = _upper(candidate.get("qspec")) + if _QSPEC_RE.match(qspec): + return qspec + return "" + + +def _intel_sspec(candidate: dict[str, Any]) -> str: + for key in ("sspec", "s_spec"): + sspec = _upper(candidate.get(key)) + if _SSPEC_RE.match(sspec): + return sspec + return "" + + +def _amd_opn(candidate: dict[str, Any]) -> str: + return _norm(candidate.get("opn")).replace(" ", "") + + +def _clock_far_below_retail(candidate: dict[str, Any]) -> bool: + base = candidate.get("base_clock_ghz") + retail = candidate.get("retail_base_clock_ghz") + if not isinstance(base, (int, float)) or isinstance(base, bool): + return False + if isinstance(retail, (int, float)) and not isinstance(retail, bool): + return base <= retail * 0.6 + # Early Intel ES desktop parts often ship a ~1.x GHz fuse default. + return base <= 1.5 + + +def _clocks_match_retail(candidate: dict[str, Any]) -> bool: + base = candidate.get("base_clock_ghz") + retail = candidate.get("retail_base_clock_ghz") + if not isinstance(base, (int, float)) or not isinstance(retail, (int, float)): + return False + if isinstance(base, bool) or isinstance(retail, bool): + return False + return abs(base - retail) <= 0.15 + + +def _classify_intel(candidate: dict[str, Any], blob: str) -> Classification: + sspec = _intel_sspec(candidate) + if sspec: + return Classification( + CLASS_RETAIL, (f"Intel S-spec {sspec} is a production part",) + ) + + qspec = _intel_qspec(candidate) + qs_phrase = _has_phrase(blob, _QS_PHRASES) + es_phrase = _has_phrase(blob, _ES_PHRASES) + stepping = _upper(candidate.get("stepping")) + + if not qspec: + if _has_phrase(blob, _RETAIL_PHRASES) or ( + "core i" in blob and "sample" not in blob + ): + return Classification( + CLASS_RETAIL, ("Intel retail model name with no Q-spec",) + ) + return Classification(CLASS_UNKNOWN, ("Intel candidate has no Q-spec or S-spec",)) + + reasons: list[str] = [f"Intel Q-spec {qspec}"] + es_hits = 0 + qs_hits = 0 + + if stepping in _ES_STEPPINGS: + es_hits += 1 + reasons.append(f"early stepping {stepping}") + if stepping in _QS_STEPPINGS: + qs_hits += 1 + reasons.append(f"mature stepping {stepping}") + if _clock_far_below_retail(candidate): + es_hits += 1 + reasons.append("base clock far below retail") + if _clocks_match_retail(candidate): + qs_hits += 1 + reasons.append("clocks match retail equivalent") + if es_phrase: + es_hits += 1 + reasons.append(f"source text {es_phrase!r}") + if qs_phrase: + qs_hits += 1 + reasons.append(f"source text {qs_phrase!r}") + + if qs_hits > es_hits: + return Classification(CLASS_QS, tuple(reasons)) + if es_hits > 0 and es_hits >= qs_hits: + return Classification(CLASS_ES, tuple(reasons), sample_revision="es1") + return Classification( + CLASS_UNKNOWN, + tuple(reasons + ["not enough ES/QS evidence"]), + ) + + +def _classify_amd(candidate: dict[str, Any], blob: str) -> Classification: + opn = _amd_opn(candidate) + qs_phrase = _has_phrase(blob, _QS_PHRASES) + es_phrase = _has_phrase(blob, _ES_PHRASES) + + if qs_phrase: + return Classification( + CLASS_QS, (f"source text {qs_phrase!r}", f"opn={opn or 'n/a'}") + ) + + if opn: + compact = opn.replace("_", "-") + if _AMD_MODERN_OPN_RE.match(opn): + reasons = [f"AMD modern OPN {opn}"] + if es_phrase: + reasons.append(f"source text {es_phrase!r}") + return Classification(CLASS_ES, tuple(reasons), sample_revision="es1") + prefix = compact[:1].upper() + if prefix == "Z": + return Classification(CLASS_QS, (f"AMD OPN prefix Z ({opn})",)) + if prefix == "1": + return Classification( + CLASS_ES, (f"AMD historical ES1 OPN {opn}",), sample_revision="es1" + ) + if prefix == "2": + return Classification( + CLASS_ES, (f"AMD historical ES2 OPN {opn}",), sample_revision="es2" + ) + + if es_phrase: + return Classification( + CLASS_ES, (f"source text {es_phrase!r}",), sample_revision="es1" + ) + if opn: + return Classification(CLASS_UNKNOWN, (f"AMD OPN {opn} has no ES/QS evidence",)) + return Classification(CLASS_UNKNOWN, ("AMD candidate has no OPN",)) + + +def classify(candidate: dict[str, Any]) -> Classification: + """Return es / qs / retail / unknown for one candidate dict.""" + blob = _blob(candidate) + manufacturer = _norm(candidate.get("manufacturer")).lower() + + declared = _norm(candidate.get("sample_class")).lower() + if declared in {CLASS_QS, "qualification"}: + return Classification(CLASS_QS, ("declared sample_class is qs",)) + if declared in {CLASS_RETAIL, "production"}: + return Classification(CLASS_RETAIL, ("declared sample_class is retail/production",)) + + if manufacturer == "intel": + return _classify_intel(candidate, blob) + if manufacturer == "amd": + return _classify_amd(candidate, blob) + + if _has_phrase(blob, _QS_PHRASES): + return Classification(CLASS_QS, ("source text indicates QS",)) + if _has_phrase(blob, _ES_PHRASES): + return Classification(CLASS_ES, ("source text indicates ES",), sample_revision="es1") + return Classification( + CLASS_UNKNOWN, (f"unsupported manufacturer '{manufacturer or 'missing'}'",) + ) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + description="Classify a CPU candidate as es, qs, retail, or unknown." + ) + parser.add_argument( + "path", + nargs="?", + help="JSON file (object or list of objects). Reads stdin if omitted.", + ) + args = parser.parse_args(argv) + + if args.path: + raw = Path(args.path).read_text(encoding="utf-8-sig") + else: + raw = sys.stdin.read() + payload: Any = json.loads(raw) + items = payload if isinstance(payload, list) else [payload] + results = [classify(item).to_dict() for item in items] + json.dump(results if isinstance(payload, list) else results[0], sys.stdout, indent=2) + sys.stdout.write("\n") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_classify.py b/tests/test_classify.py new file mode 100644 index 0000000..313623f --- /dev/null +++ b/tests/test_classify.py @@ -0,0 +1,128 @@ +"""Known public identifiers: QXLB=es, QDF4=qs, 12900K=retail, Raphael OPN=es.""" + +from __future__ import annotations + +from app.ingest.classify import classify + + +def test_intel_qxlb_is_es() -> None: + result = classify( + { + "manufacturer": "intel", + "qspec": "QXLB", + "stepping": "B0", + "base_clock_ghz": 1.2, + "retail_base_clock_ghz": 3.2, + "name": "Intel QXLB Engineering Sample", + "markings": ["Intel Confidential"], + } + ) + assert result.sample_class == "es" + assert result.sample_revision == "es1" + + +def test_intel_qdf4_is_qs() -> None: + result = classify( + { + "manufacturer": "intel", + "qspec": "QDF4", + "stepping": "C0", + "base_clock_ghz": 3.2, + "retail_base_clock_ghz": 3.2, + "name": "i9-14900K QS", + "source_labels": ["qs"], + } + ) + assert result.sample_class == "qs" + + +def test_intel_s_spec_is_retail() -> None: + result = classify( + { + "manufacturer": "intel", + "sspec": "SRKNY", + "name": "Intel Core i9-14900K", + } + ) + assert result.sample_class == "retail" + + +def test_intel_retail_model_without_qspec() -> None: + result = classify( + { + "manufacturer": "intel", + "name": "Intel Core i9-12900K", + "slug": "core-i9-12900k", + } + ) + assert result.sample_class == "retail" + + +def test_intel_qspec_without_evidence_is_unknown() -> None: + result = classify({"manufacturer": "intel", "qspec": "QZZZ"}) + assert result.sample_class == "unknown" + + +def test_amd_raphael_opn_is_es() -> None: + result = classify( + { + "manufacturer": "amd", + "opn": "100-000000665-21_N", + "name": "AMD Eng Sample 100-000000665-21_N", + } + ) + assert result.sample_class == "es" + + +def test_amd_z_prefix_is_qs() -> None: + result = classify( + { + "manufacturer": "amd", + "opn": "ZS188159TGG54", + "name": "AMD qualification sample", + } + ) + assert result.sample_class == "qs" + + +def test_amd_historical_es1_prefix() -> None: + result = classify({"manufacturer": "amd", "opn": "1S160805L4BGC"}) + assert result.sample_class == "es" + assert result.sample_revision == "es1" + + +def test_amd_historical_es2_prefix() -> None: + result = classify({"manufacturer": "amd", "opn": "2S160805L4BGC"}) + assert result.sample_class == "es" + assert result.sample_revision == "es2" + + +def test_declared_qs_wins() -> None: + result = classify( + { + "manufacturer": "intel", + "qspec": "QXLB", + "sample_class": "qs", + } + ) + assert result.sample_class == "qs" + + +def test_declared_retail_wins() -> None: + result = classify( + { + "manufacturer": "amd", + "opn": "100-000000665-21_N", + "sample_class": "retail", + } + ) + assert result.sample_class == "retail" + + +def test_seed_qxlb_file_classifies_es() -> None: + from pathlib import Path + import json + + path = Path("data/cpu/intel/2021/desktop/intel-qxlb.json") + rec = json.loads(path.read_text(encoding="utf-8")) + assert classify(rec).sample_class == "es"