Any field that accepts image uploads is an open invitation for explicit content. Of course, not everyone will interpret your upload fields as such — but unfortunately, as ever, we must to march to the beat of the most erratic and unpredictable user.

It’s a well-established fact that eyeballing every upload doesn’t scale, so we’ll of course need an automatic score to gate uploads on. Thankfully, in the bustling age of AI, there’s no shortage of powerful solutions we can count on to provide this exact service.

Here, we’ll handle explicit content detection via the Cloudmersive Image Recognition API’s Python client.

Installing and configuring the client

Let’s begin by installing the latest version of the Image API client:

pip install cloudmersive-image-api-client

And let’s now import resources & configure the client using our Cloudmersive API key:

import os
import cloudmersive_image_api_client

config = cloudmersive_image_api_client.Configuration()
config.api_key["Apikey"] = os.environ["CLOUDMERSIVE_API_KEY"]

nsfw_api = cloudmersive_image_api_client.NsfwApi(
    cloudmersive_image_api_client.ApiClient(config)
)

Missing an API key? You can get one for free by creating a free account on the Cloudmersive website (this allows up to 800 API calls per month with zero commitments). It must be said, however, that explicit content detection is a resource-intensive operation, so you might consider small business or enterprise-tier keys if you’re dealing with a high-volume workflow.

Scoring an image (defining its explicitness)

Through this API, we’re working with a classier that returns a score from 0.0 (clearly safe) to 1.0 (clearly explicit), with the racy middle ground living inbetween:

def nsfw_score(path):
    result = nsfw_api.nsfw_classify(path)
    return result.score

Deciding what to allow based on the score

To use this API effectively, we need to pick a threshold that matches our own specific tolerance. As a rough guide, we can consider anything under 0.2 as safe, around 0.7 as racy, and above 0.8 as explicit; so 0.5 is a reasonably strict cutoff to stay on the safe side.

def is_allowed(path, threshold=0.5):
    return nsfw_score(path) < threshold

if not is_allowed("uploads/avatar.jpg"):
    reject("This image was flagged as inappropriate.")

Conclusion

Yep, we’re already done — that’s all there is to it. Just plug that into your workflow and let automated decisions filter out the bad stuff. If you find that too much borderline content is being blocked, all you need to do is adjust your score threshold in the minimal code we’ve written, and you’re all set.

You’ve got a list of uploaded documents, and you want a small visual preview beside each one (the way a file browser shows them). Good choice.

The only trouble is that rendering a Word doc or PDF to an image yourself means wrangling a headless renderer for every format your platform accepts. That’s not worth the trouble.

Thankfully, there’s an easier way: we’ll generate PNG thumbnails with a few lines of code using the Cloudmersive Convert API.

Installing & configuring the client

First, let’s install the client with the below pip command:

pip install cloudmersive-convert-api-client

Next, let’s import the resources we need & configure our API key:

import os
import cloudmersive_convert_api_client

config = cloudmersive_convert_api_client.Configuration()
config.api_key["Apikey"] = os.environ["CLOUDMERSIVE_API_KEY"]

convert_api = cloudmersive_convert_api_client.ConvertDocumentApi(
    cloudmersive_convert_api_client.ApiClient(config)
)

We can get free API key by signing up for a free account on the Cloudmersive website; that’ll give us a limit of 800 API calls/month with no commitments.

Creating the thumbnail

The “autodetect thumbnail” endpoint we’ll be using figures out the document’s format on its own and returns a PNG as bytes. You just have to set the maximum dimensions; it scales to fit automatically:

def make_thumbnail(path, width=128, height=128):
    return convert_api.convert_document_autodetect_to_thumbnail(
        path, max_width=width, max_height=height
    )

Saving the thumbnail

This part’s just as easy. We’ll use a quick helper to handle saving:

def save_thumbnail(path, out_path, width=128, height=128):
    png = make_thumbnail(path, width, height)
    with open(out_path, "wb") as f:
        f.write(png)

save_thumbnail("uploads/report.docx", "thumbs/report.png")

A Word doc, a PDF, a spreadsheet, etc. — each comes back as a ready-to-display PNG at the size you specified.

Conclusion

One call turns any uploaded document into a preview image with no per-format rendering on your side. Just point this at whever users send through and get back a thumbnail you can drop straight into your UI.

People like spreadsheets, but your app likes structured data. The trouble is that parsing XLSX for your app with custom code is no fun; it’s fiddly, and it can break the moment an Excel sheet defies our expectations.

It’s ultimately far less work to convert spreadsheets to JSON in one quick API call, and then tidy the result.

In our unbiased opinion, it’s best to handle this in Python using the Cloudmersive Convert API client. Let’s walk through that now.

Installing and configuring the client

To begin, let’s run the following pip command to install the client:

pip install cloudmersive-convert-api-client

After that, let’s configure the client with our API key:

import os
import cloudmersive_convert_api_client

config = cloudmersive_convert_api_client.Configuration()
config.api_key["Apikey"] = os.environ["CLOUDMERSIVE_API_KEY"]

data_api = cloudmersive_convert_api_client.ConvertDataApi(
    cloudmersive_convert_api_client.ApiClient(config)
)

We can get a free API key by creating a free account on the Cloudmersive website (this allows 800 API calls/month with no commitments).

Converting the spreadsheet to JSON

The XLSX-to-JSON endpoint we’ll be using reads the sheet and hands back a row of objects, each keyed by the column headers. Ther’es no cell math required.

def xlsx_to_rows(path):
    return data_api.convert_data_xlsx_to_json(path)

Cleaning the output

We might run into an issue of spreadsheets carrying blank rows and stray whitespace — so let’s include a pass that trims strings and drops rows that are entirely empty:

def clean(rows):
    cleaned = []
    for row in rows:
        row = {k: (v.strip() if isinstance(v, str) else v) for k, v in row.items()}
        if any(v not in (None, "") for v in row.values()):
            cleaned.append(row)
    return cleaned

Putting it together

def load_spreadsheet(path):
    return clean(xlsx_to_rows(path))

for record in load_spreadsheet("uploads/orders.xlsx"):
    print(record)

Now, an uploaded workbook comes out as a clean list of dictionaries, ready to validate, store, or hand to the next step in our pipeline.

Conclusion

In one quick API call, we managed to turn spreadsheets into rows of objects — exactly what our application likes to digest. No manual parsing required.

For any signup page, a way-too-large share of the emails that come in are junk. Think typo’d domains, addresses that don’t resolve, or throwaway inboxes from disposable-mail services that *poof* vanish in ten minutes.

Let’s avoid regex and instead fix that by adding a quick and easy email validation service to our signup processing workflow. We’ll use the Cloudmersive Validate API’s email_full_validation() method to check an email address’s syntax, domain, mail server, and disposable status in a single call.

Installing and configuring the client

To install the client, let’s run this command:

pip install cloudmersive-validate-api-client

And following that, let’s configure the client:

import os
import cloudmersive_validate_api_client

config = cloudmersive_validate_api_client.Configuration()
config.api_key["Apikey"] = os.environ["CLOUDMERSIVE_API_KEY"]

email_api = cloudmersive_validate_api_client.EmailApi(
    cloudmersive_validate_api_client.ApiClient(config)
)

We’ll need to replace “CLOUDMERSIVE_API_KEY” with our actual API key string. To get one of those, we can create a free account on the Cloudmersive website; we’ll get 800 API calls/month with no commitments.

Validating an email address

Full email validation goes far beyond checking syntax. It’s about confirming the domain actually exists, that the mail server actually accepts mail, and whether the domain is a known disposable provider.

def check_email(address):
    request = cloudmersive_validate_api_client.FullEmailValidationRequest(email=address)
    return email_api.email_full_validation(request)

Deciding at signup

Thgouth valid_address, we get syntax, domain, and SMTP in one verdict. We can pair that with is_dispoable to turn away burner inboxes even when they’re technically deliverable.

def is_acceptable(address):
    result = check_email(address)
    return result.valid_address is True and result.is_disposable is not True

if not is_acceptable(form_email):
    reject("Please use a valid, non-disposable email address.")

Conclusion

Checking deliverability and disposability at signup keeps your user table full of email addresses you can actually reach. That’s far more valuable than a table full of bounces and throwaway accounts (you’ll just be scrubbing those out later, and no one wants to do that).

If you’re handling user input in Python, you laready know the standard defenses: parameterized queries (or an ORM) to neutralize SQL injection, and output encoding + content security policy to shut down XSS. Those are the right tools, and nothing we’ll discuss here replaces them; they stop the attack at the sink.

But it’s worth noting these defenses stop attacks silently. There’s no built-in threat logging mechanism that comes with implementing “best practices”. Parameterized queries turn a malicious string into a harmless literal and move on; in other words, you never actually learn that you were probed, by whom, or how often that entity was probing.

That blind spot is where a bona fide threat detection service earns its place. Screening input at the boundary gives you the signal your sink-level defenses throw away: an attempt you can log, rate-limit, alert on, or block before it reaches code paths your escaping doesn’t cover (logs, downstream services, data stored now and rendered somewhere else later).

Let’s add that advanced detection layer in Python using the Cloudmersive Security API.

Installing the client

Let’s first run this command to install the client:

pip install cloudmersive-security-api-client

Configuring the client

Next up, let’s import the resources and configure:

import os
import cloudmersive_security_api_client

config = cloudmersive_security_api_client.Configuration()
config.api_key["Apikey"] = os.environ["CLOUDMERSIVE_API_KEY"]

threat_api = cloudmersive_security_api_client.ContentThreatDetectionApi(
    cloudmersive_security_api_client.ApiClient(config)
)

We can create a connection with a free API key by first creating a free account on the Cloudmersive website.

Checking each threat

We’ll make exactly one API call per threat type — each reduced to a simple boolean:

def has_sql_injection(text):
    result = threat_api.content_threat_detection_check_sql_injection_string(text)
    return result.contained_sql_injection_attack is True

def has_xss(text):
    result = threat_api.content_threat_detection_protect_xss(text)
    return result.contained_xss is True

Screening an input

Now we’ll combine them into one pass that reports which threats (if any) it found:

def screen_input(text):
    threats = []
    if has_sql_injection(text):
        threats.append("sql_injection")
    if has_xss(text):
        threats.append("xss")
    return threats

threats = screen_input(request_field)
if threats:
    raise ValueError(f"Rejected input: {', '.join(threats)}")

If you’d rather cover both threats (and several more) in a single request, the API’s automatic threat detection endpoint scans for multiple threat types at once (content_threat_detection_automatic_threat_detection_string()).

Conclusion

Our own detection logic keeps attacks from landing, but true detection tells you it was attempted. Running this screen at the application’s input boundary turns silent protection into something you can see, log, and respond to.

Nothing stops someone from renaming payload.exe to invoice.pdf. To actually trust a file’s type, you have to read its contents, not just its name.

Writing software that reads file bytes is both boring and complicated. It’s also a pain to update. So let’s use an API instead — specifically, the Cloudmersive Convert API, which exposes methods for detecting a file’s real format no matter what the filename claims.

Installing the client

First things first, let’s install the Cloudmersive client with the below command:

pip install cloudmersive-convert-api-client

Configuring the client

Next, let’s import and configure the client:

import os
import cloudmersive_convert_api_client

config = cloudmersive_convert_api_client.Configuration()
config.api_key["Apikey"] = os.environ["CLOUDMERSIVE_API_KEY"]

convert_api = cloudmersive_convert_api_client.ConvertDocumentApi(
    cloudmersive_convert_api_client.ApiClient(config)
)

Detecting the real type

The autodetect endpoint inspects the file’s contents and returnst he extension it actually matches (complete with the leading period):

def detected_extension(path):
    info = convert_api.convert_document_autodetect_get_info(path)
    return info.detected_file_extension   # e.g. ".pdf", ".docx", ".png"

Comparing claimed against actual

Now we can simply line up the extension on the filename against the one the file contents reveal. If we get a mismatch, we’re dealing with a spoofed file.

import pathlib

def verify_type(path):
    claimed = pathlib.Path(path).suffix.lower()
    actual = (detected_extension(path) or "").lower()

    return {"claimed": claimed, "actual": actual, "matches": claimed == actual}

result = verify_type("uploads/invoice.pdf")
if not result["matches"]:
    print(f"Rejected: claims {result['claimed']} but is really {result['actual']}")Conclusion

With this logic in place, a genuine PDF passes, while an executable wearing a .pdf suffix comes back as .exe and gets rejected before it can do end up somewhere it shouldn’t.

Conclusion

File contents are the truth, with the filename is just a claim. Verifying detected file types against claimed extensions holds firm against surface-level spoofing attempts.

File scanners typically give us one of two outcomes: clean, or not clean. But in real document workflows, “not clean” is often too blunt. A password-protected PDf or a macro-enabled spreadsheet will trip a strict scanner, yet both are routine business documents. Blocking them outright frustrates users; waving them through anyway defeats the purpose of the scan.

If we’re stuck dealing with files like these in our workflow, the best bet is to send the ambiguous files to a human reviewer.

It’s easy enough to automate that process. We’ll build a three-way router in Python using an Enum for outcomes, a dataclass to carry the decision, and structural pattern matching (Python 3.10+) to dispath on it. As usual, we’ll handle the actual scanning portion of our workflow with the Cloudmersive Advanced Scan API, whose per-category flags let us tell “dangerous” files apart from “work a second look” files.

Installing and configuring

Let’s first install Cloudmersive:

pip install cloudmersive-virus-api-client

And then let’s configure our API key:

import os
import cloudmersive_virus_api_client

config = cloudmersive_virus_api_client.Configuration()
config.api_key["Apikey"] = os.environ["CLOUDMERSIVE_API_KEY"]

scan_api = cloudmersive_virus_api_client.ScanApi(
    cloudmersive_virus_api_client.ApiClient(config)
)

Modeling the decision

Now let’s model three outcomes, and build a decision that carries the reasons behind it:

from enum import Enum
from dataclasses import dataclass, field

class Outcome(Enum):
    CLEAN = "clean"
    REVIEW = "review"
    BLOCK = "block"

@dataclass
class Decision:
    outcome: Outcome
    reasons: list[str] = field(default_factory=list)

Triaging a scan result

The point is that some findings are unambiguously hostile, while other just warrant a closer look. We’ll split the advanced scan’s flags into those two groups and decide accordingly:

HARD_THREATS = {
    "contains_executable": "executable content",
    "contains_script": "embedded script",
    "contains_insecure_deserialization": "insecure deserialization",
    "contains_xml_external_entities": "XML external entity",
}
SOFT_FLAGS = {
    "contains_macros": "macros",
    "contains_password_protected_file": "password protection",
    "contains_ole_embedded_object": "embedded OLE object",
}

def triage(result) -> Decision:
    if result.found_viruses:
        return Decision(Outcome.BLOCK, [v.virus_name for v in result.found_viruses])

    hard = [label for attr, label in HARD_THREATS.items() if getattr(result, attr)]
    if hard:
        return Decision(Outcome.BLOCK, hard)

    if result.clean_result is True:
        return Decision(Outcome.CLEAN)

    soft = [label for attr, label in SOFT_FLAGS.items() if getattr(result, attr)]
    return Decision(Outcome.REVIEW, soft or ["flagged, reason unspecified"])

Routing the document

Now, the scan runs, triage decides, and match sends the file down the right path:

def route(path, review_queue: list[str]) -> Decision:
    result = scan_api.scan_file_advanced(
        path,
        allow_executables=False,
        allow_scripts=False,
        allow_macros=False,
        allow_invalid_files=False,
        allow_password_protected_files=False,
        allow_unsafe_archives=False,
    )
    decision = triage(result)

    match decision.outcome:
        case Outcome.CLEAN:
            pass                      # hand off to normal processing
        case Outcome.REVIEW:
            review_queue.append(path) # a human looks before it proceeds
        case Outcome.BLOCK:
            pass                      # drop it; log decision.reasons

    return decision

Note thatreview_queue is a list here, but in a real-world context it’s the seam where you’d push onto a SQS, a database table, or a Celery task in production.

Conclusion

Reading the individual scan flags instead of one clean/not-clean verdict lets you keep dangerous files out of your system while giving borderline ones the human judgement they actually need. The triage rules live in one place, so re-drawing the line between block and review is just a one-line change.

You’re accepting document uploads, and you need them all as PDF (for previews, for archiving, or simply to give the rest of your system one predictable format to work with).

The conversion part is easy; the uncomfortable part is that you’re feeding files you didn’t create into a converter. A malformed or malicious upload is exactly the kind of input you’d rather that converter (or anything in the software chain) never touched.

So when we talk about “safe” PDF conversion, it boils down to one rule: screen every file before converting it.

In the below example, we’ll run each upload through the Cloudmersive Virus Scan API before handing clean documents to a PDF conversion API.

Installing the clients

pip install cloudmersive-convert-api-client cloudmersive-virus-api-client

Configuring the clients

Each of the above packages carries its own configuration, so we’ll set both from the same Cloudmersive API key:

import os
import cloudmersive_convert_api_client
import cloudmersive_virus_api_client

api_key = os.environ["CLOUDMERSIVE_API_KEY"]

convert_config = cloudmersive_convert_api_client.Configuration()
convert_config.api_key["Apikey"] = api_key

scan_config = cloudmersive_virus_api_client.Configuration()
scan_config.api_key["Apikey"] = api_key

convert_api = cloudmersive_convert_api_client.ConvertDocumentApi(
    cloudmersive_convert_api_client.ApiClient(convert_config)
)
scan_api = cloudmersive_virus_api_client.ScanApi(
    cloudmersive_virus_api_client.ApiClient(scan_config)
)

Screening a file

The advanced scan method from the Virus Scan API rejects the exact type of content we never want near a converter: executables, scripts, macros, malformed fils, unsafe archives, and others. The below helper reduces the API response to a simple “yes” or “no” gate pre-conversion:

def passes_screening(path):
    result = scan_api.scan_file_advanced(
        path,
        allow_executables=False,
        allow_scripts=False,
        allow_macros=False,
        allow_invalid_files=False,
        allow_password_protected_files=False,
        allow_unsafe_archives=False,
    )
    return result.clean_result is True

Converting to PDF

For the conversion itself, the “autodetect and convert” endpoint we’re using figures out the input format on its own, which is useful when users can send Word, Excel, PowerPoint images, and other similarly common file types through the same path. It hands back the new PDF document as bytes:

def to_pdf(path):
    return convert_api.convert_document_autodetect_to_pdf(path)

Putting it together

Now we’ll assemble our code in the order that makes our conversion safe. We’ll screen first, and convert only on a “pass”. A file that fails its threat check never reaches to_pdf:

def convert_safely(path, output_path):
    if not passes_screening(path):
        raise ValueError(f"{path} failed screening; conversion skipped.")

    pdf_bytes = to_pdf(path)
    with open(output_path, "wb") as f:
        f.write(pdf_bytes)

convert_safely("uploads/report.docx", "converted/report.pdf")

A clean document comes out the other side as a PDF on disk, while a suspicious one raises before the converter is hever called.

Conclusion

Since our scan gates our conversion directly, the converter only ever works on input we’ve thoroughly vetted. Importantly, the screening and conversion steps also stay independent, so we can tighten the rules or change the output format without disturbing the other half.

For a handful of files, looping API calls over a folder works fine. For a few hundred files, however, we’ll hit rate limits or trigger timeouts pretty quickly, and we might run up an unepxected bill. We have to remember that Python lets us fire off requests really fast.

We don’t need to slow the whole script down to solve the problem. Rather, we can limit how many reqeusts are ever in flight at once.

Below, we’ll learn how to implement that logic in Python before making API calls to the Cloudmersive Document AI Field Extraction API.

Installing the client

pip install cloudmersive-documentai-api-client

Configuring Document AI

import os
import cloudmersive_documentai_api_client
from cloudmersive_documentai_api_client.rest import ApiException

config = cloudmersive_documentai_api_client.Configuration()
config.api_key["Apikey"] = os.environ["CLOUDMERSIVE_API_KEY"]

extract_api = cloudmersive_documentai_api_client.ExtractApi(
    cloudmersive_documentai_api_client.ApiClient(config)
)

FIELDS = "Invoice Number,Invoice Date,Total Due"

Extracting from a single file

In the below code, we’ll isoltate the work for one document it its own function. Returning errors instead of raising them means a single bad file won’t sink the whole batch:

def extract_one(path):
    try:
        response = extract_api.extract_fields(
            field_names=FIELDS,
            recognition_mode="Advanced",
            input_file=str(path),
        )
        return path.name, {r.field_name: r.field_string_value for r in response.results}
    except ApiException as error:
        return path.name, {"error": str(error)}

Capping how many run at once

Here, we’ll keep the API from getting overwhelmed. A ThreadPoolExecutor with a small max_workers means only that specific number of requests are ever active at a time, no matter how many files the original folder holds. The moment a slot frees up, the pool picks up the next file.

import pathlib
from concurrent.futures import ThreadPoolExecutor

def process_folder(folder, max_workers=4):
    files = [p for p in pathlib.Path(folder).iterdir() if p.is_file()]

    results = {}
    with ThreadPoolExecutor(max_workers=max_workers) as pool:
        for name, fields in pool.map(extract_one, files):
            results[name] = fields

    return results

Running the batch

summary = process_folder("./invoices", max_workers=4)

for name, fields in summary.items():
    print(name, fields)

Now four files process concurrently, while the other 496 patiently wait their turn. If the API enforces a strict per-second rate rather than a concurrency limit, we can simply lower max_workers futher, or possibly add a short time.sleep inside extract_one to space the calls out.

Conclusion

Whether our starting folder holds five documents or five thousand, max_workers is what ultimately decides how hard we lean on the API. Tuning that single number is how we balance throughput against API limits.

While it’s true that we could scan files with an S3 trigger after upload, that would still let a bad file land in our bucket first — and we don’t want that. Scanning before the upload means the file never arrives at all, and that’s a much safer bet.

Let’s handle that with a pre-upload gate. We’ll use boto3 and the Cloudmersive Advanced Virus Scan API.

Installing the dependencies

pip install boto3 cloudmersive-virus-api-client

Connecting to boto3 and Cloudmersive

import os
import boto3
import cloudmersive_virus_api_client

cfg = cloudmersive_virus_api_client.Configuration()
cfg.api_key["Apikey"] = os.environ["CLOUDMERSIVE_API_KEY"]

scan_api = cloudmersive_virus_api_client.ScanApi(
    cloudmersive_virus_api_client.ApiClient(cfg)
)

s3 = boto3.client("s3")
BUCKET = "my-upload-bucket"

Note that we can get a free Cloudmersive API key by creating a free account on the Cloudmersive website.

Building a gate for a single file

Since these files already sit on disk, we can scan each file path directly. We’ll just put the whole decision (scan → upload || refuse) into one function that reports back what it did.

import pathlib

def upload_if_clean(path: pathlib.Path) -> str:
    verdict = scan_api.scan_file_advanced(
        str(path),
        allow_executables=False,
        allow_scripts=False,
        allow_macros=False,
        allow_invalid_files=False,
        allow_password_protected_files=False,
        allow_unsafe_archives=False,
    )

    if verdict.clean_result is not True:
        return "blocked"

    s3.upload_file(str(path), BUCKET, path.name)
    return "uploaded"p

Running it over a folder

Now that we’ve implemented per-file decision making, processing a directory is a short loop. Each file is responsible for announcing its own outcome:

for item in pathlib.Path("./outbox").iterdir():
    if item.is_file():
        print(f"{item.name}: {upload_if_clean(item)}")

A blocked file simply can’t land in the bucket because it never reaches upload_file.

Wrapping up

Vetting files in this simple, easy way keeps our bucket safe by design instead of relying on a cleanup pass later. The same gate works against any bucket or prefix we aim it at because the scan and transfer stay decoupled.