TEXT_COL = "skill_md_content"
NUM_COLS = ["skillspector_score", "static_finding_count",
"skillspector_issue_count", "virustotal_malicious_count"]
TARGET = "clawscan_verdict"
def prep(df):
out = df.copy()
out[TEXT_COL] = out[TEXT_COL].fillna("").astype(str).str.slice(0, 6000)
for c in NUM_COLS:
out[c] = pd.to_numeric(out[c], errors="coerce")
return out
train_p, test_p = prep(train_df), prep(test_df)
get_text = FunctionTransformer(lambda X: X[TEXT_COL].values, validate=False)
text_pipe = Pipeline([
("select", get_text),
("tfidf", TfidfVectorizer(max_features=20000, ngram_range=(1,2),
min_df=3, sublinear_tf=True)),
])
num_pipe = Pipeline([
("impute", SimpleImputer(strategy="constant", fill_value=0)),
("scale", StandardScaler()),
])
features = ColumnTransformer([
("text", text_pipe, [TEXT_COL]),
("num", num_pipe, NUM_COLS),
])
clf = Pipeline([
("features", features),
("model", LogisticRegression(max_iter=2000, C=4.0,
class_weight="balanced",
multi_class="multinomial")),
])
print("\nTraining classifier (SKILL.md text + scanner numbers -> verdict)...")
clf.fit(train_p[[TEXT_COL] + NUM_COLS], train_p[TARGET])
pred = clf.predict(test_p[[TEXT_COL] + NUM_COLS])
print("\n=== Test-set classification report ===")
print(classification_report(test_p[TARGET], pred, digits=3))
cm = confusion_matrix(test_p[TARGET], pred, labels=order)
plt.figure(figsize=(6,5))
sns.heatmap(cm, annot=True, fmt="d", cmap="Blues", xticklabels=order, yticklabels=order)
plt.title("Confusion matrix (test split)"); plt.xlabel("Predicted"); plt.ylabel("Actual"); plt.show()
test_out = test_p[["skill_slug", TARGET, "clawscan_summary"]].copy()
test_out["pred"] = pred
errors = test_out[test_out[TARGET] != test_out["pred"]].head(8)
print("\n=== Sample misclassifications ===")
for _, r in errors.iterrows():
print(f"- {r['skill_slug']:35s} true={r[TARGET]:10s} pred={r['pred']:10s}")
print("\nDone. Set SAMPLE_SIZE=None for the full dataset.")
