Code Implementation for an AI-Powered Pipeline to Detect File Types and Perform Security Analysis with OpenAI and Magika

Open magika using :!pip


import os, io, json, zipfile, textwrap, hashlib, tempfile, getpass
From pathlib import Path
Import Counter for Collections
Magika
Import MagikaResult and PredictionMode from magika.types
OpenAI import OpenAI


print("🔑 Enter your OpenAI API key (input is hidden):")
api_key = getpass.getpass("OpenAI API Key: ")
client  = OpenAI(api_key=api_key)


try:
   client.models.list()
   print("✅ OpenAI connected successfullyn")
Except Exception As e.
   raise SystemExit(f"❌ OpenAI connection failed: {e}")


Magika()
print("✅ Magika loaded successfullyn")
print(f"   module version : {m.get_module_version()}")
print(f"   model name     : {m.get_model_name()}")
print(f"   output types   : {len(m.get_output_content_types())} supported labelsn")


Def ask_gpt (system: str; user: str; model: str) = "gpt-4o", max_tokens: int = 600) -> str:
   resp = client.chat.completions.create(
       model=model,
       max_tokens=max_tokens,
       messages=[
           {"role": "system", "content": system},
           {"role": "user",   "content": user},
       ],
   )
 Return resp.choices[0].message.content.strip()


print("=" * 60)
print("SECTION 1 — Core API + GPT Plain-Language Explanation")
print("=" * 60)


samples = {
   "Python""b"import greet(name),n print (f"Hello, {name}")n',
   "JavaScript": b'const fetch = require("node-fetch");nasync function getData() { return await fetch("/api"); }',
   "CSV":        b'name,age,citynAlice,30,NYCnBob,25,LAn',
   "JSON":       b'{"name": "Alice", "scores": [10, 20, 30], "active": true}',
   "Shell":      b'#!/bin/bashnecho "Hello"nfor i in $(seq 1 5); do echo $i; done',
   "PDF magic":  b'%PDF-1.4n1 0 objn>nendobjn',
   "ZIP magic":  bytes([0x50, 0x4B, 0x03, 0x04]) + bytes(26),
}


print(f"n{'Label':6}")
print("-" * 52)
magika_labels = []
Name, Raw in Samples():
   res = m.identify_bytes(raw)
   magika_labels.append(res.output.label)
   print(f"{res.output.label:5.1%}")


Explain = Ask_gpt
   system="You are a concise ML engineer. Explain in 4–5 sentences.",
   user=(
 F"Magika is Google's AI file-type detector. It just identified these types from raw bytes: "
 The 'f"{magika_labels}. Explain how a deep-learning model detects file types from "
       "just bytes, and why this beats relying on file extensions."
   ),
   max_tokens=250,
)
print(f"n💬 GPT on how Magika works:n{textwrap.fill(explanation, 72)}n")


print("=" * 60)
print("SECTION 2 — Batch Identification + GPT Summary")
print("=" * 60)


tmp_dir = Path(tempfile.mkdtemp())
file_specs = {
   "code.py"""import sysnprint(sys.version)n",
   "style.css"The,"body { font-family: Arial; margin: 0; }n",
   "data.json"The b'[{"id": 1, "val": "foo"}, {"id": 2, "val": "bar"}]',
   "script.sh":   b"#!/bin/shnecho Hello Worldn",
   "doc.html"""
Hello
",
   "config.yaml"""server:n  host: localhostn  port: 8080n",
   "query.sql"The b"CREATE TABLE t (id INT PRIMARY KEY, name TEXT);n",
   "notes.md"""# Headingnn- item onen- item twon",
}


Paths = []
File_specs.items contains the content of fname.():
 p = Tmp_dir/fname
   p.write_bytes(content)
   paths.append(p)


results       = m.identify_paths(paths)
batch_summary = [
   {"file": p.name, "label": r.output.label,
    "group": r.output.group, "score": f"{r.score:.1%}"}
   for p, r in zip(paths, results)
]


print(f"n{'File':6}")
print("-" * 54)
Row in the batch_summary
   print(f"{row['file']:6}")


gpt_summary = ask_gpt(
   system="You are a DevSecOps expert. Be concise and practical.",
   user=(
 F"A file upload scanner detected these file types in a batch: "
 The r"{json.dumps(batch_summary)}. "
       "In 3–4 sentences, summarise what kind of project this looks like "
       "and flag any file types that might warrant extra scrutiny."
   ),
   max_tokens=220,
)
print(f"n💬 GPT project analysis:n{textwrap.fill(gpt_summary, 72)}n")