From Weekend Hack to Side Income: Python Automation with Flask
1) The small pain worth solving Your inbox floods with customer emails and bug reports. You skim, label, and forward. Repeat. It steals an hour a day. Here’s the fix: a tiny Python automation tool that turns raw text into a short summary and a label — bug, feature, or billing — via a single /classify endpoint. Ship it as a service. Point your helpdesk to it. Measure time saved.
2) TL;DR We’ll build a minimal Flask service (a FastAPI alternative) that accepts text, calls a model via litellm, and returns {label, summary}. Includes Pydantic v2 schemas, Tenacity retries, a Dockerfile, a deterministic test (MOCK mode), plus curl/httpx clients. Use it to power content AI workflows or bootstrap an AI SaaS.
3) Why this matters now
- Teams need triage, not dashboards. Labels + summaries unblock routing and response SLAs.
- Content AI = ROI. Repetitive reading is expensive; compressing text to decisions is cheap.
- Python automation scales horizontally: cron + queues now, more integrations later.
- Same bones, many jobs. This pattern adapts to invoice processing automation or PDF data extraction with OCR later; for today, we keep it lean and text-first.
4) The tiny architecture
flowchart LR
A[Client / Helpdesk] -->|POST /classify {text}| B[Flask app]
B --> C[Validator (Pydantic v2)]
C --> D{MOCK_MODE?}
D -- yes --> E[Heuristic labeler + shortener]
D -- no --> F[Model via litellm + retries]
F --> G[Parse strict JSON]
E --> H[Output]
G --> H[Output]
H --> I[(Store/Notify: Notion/Sheets/Slack)]
5) The “~55-Line” Core (Flask version)
# app.py
import os, json, re
from dotenv import load_dotenv
from flask import Flask, request, jsonify
from tenacity import retry, stop_after_attempt, wait_exponential
from schemas import InputPayload, OutputPayload
from litellm import completion # provider-agnostic wrapper
load_dotenv()
app = Flask(__name__)
MODEL = os.getenv("MODEL", "gpt-4o-mini")
API_KEY = os.getenv("API_KEY") # For OpenAI: OPENAI_API_KEY also works
if API_KEY and "OPENAI_API_KEY" not in os.environ:
os.environ["OPENAI_API_KEY"] = API_KEY # simple default for litellm
MOCK_MODE = os.getenv("MOCK_MODE", "0") == "1"
ALLOWED = {"bug", "feature", "billing"}
MAX_INPUT_CHARS = 8000
def _heuristic_label(text: str) -> str:
t = text.lower()
if any(k in t for k in ["refund", "invoice", "charge", "payment", "billing"]):
return "billing"
if any(k in t for k in ["crash", "error", "broken", "fail", "bug"]):
return "bug"
return "feature"
def _shorten(text: str, max_words: int = 40) -> str:
words = text.split()
out = " ".join(words[:max_words])
return out if out.endswith(".") else out + "."
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=0.5, max=8))
def _model_call(text: str) -> dict:
sys = (
"You are a classifier. Return STRICT JSON with keys "
'"label" (bug|feature|billing) and "summary" (<=55 words).'
)
user = f"Text:\n{text}\nRespond with ONLY JSON."
resp = completion(model=MODEL, messages=[{"role":"system","content":sys},
{"role":"user","content":user}])
content = resp["choices"][0]["message"]["content"]
# Try to parse robustly
m = re.search(r"\{.*\}", content, re.S)
return json.loads(m.group(0) if m else content)
@app.get("/health")
def health():
return {"ok": True}
@app.post("/classify")
def classify():
# Validate & guardrails
try:
payload = InputPayload(**request.get_json(force=True))
except Exception as e:
return jsonify({"error": f"Invalid payload: {e}"}), 400
text = payload.text.strip()
if not text:
return jsonify({"error":"text is required"}), 400
if len(text) > MAX_INPUT_CHARS:
return jsonify({"error":f"text too long (>{MAX_INPUT_CHARS})"}), 413
if MOCK_MODE:
label = _heuristic_label(text)
summary = _shorten(text)
return jsonify(OutputPayload(label=label, summary=summary).model_dump())
try:
data = _model_call(text)
label = (data.get("label") or "").lower().strip()
summary = (data.get("summary") or "").strip()
if label not in ALLOWED:
label = _heuristic_label(text)
if not summary:
summary = _shorten(text)
return jsonify(OutputPayload(label=label, summary=summary).model_dump())
except Exception as e:
# Last-resort fallback
return jsonify(OutputPayload(label=_heuristic_label(text),
summary=_shorten(text)).model_dump() | {
"note": f"fallback: {str(e)[:120]}"
}), 200
if __name__ == "__main__":
app.run(host="0.0.0.0", port=int(os.getenv("PORT", "8000")))
Schemas (Pydantic v2):
# schemas.py
from pydantic import BaseModel, Field
class InputPayload(BaseModel):
text: str = Field(..., description="Raw text to summarize and classify")
meta: dict | None = None
class OutputPayload(BaseModel):
label: str = Field(..., description="bug | feature | billing")
summary: str
6) Run it locally (and in Docker)
* requirements.txt
flask==3.0.3
pydantic>=2.7
httpx==0.27.2
tenacity==8.5.0
python-dotenv==1.0.1
litellm==1.45.12
pytest==8.3.2
* .env.example
# Choose a model supported by litellm (e.g., OpenAI, Anthropic, etc.)
MODEL=gpt-4o-mini
# For OpenAI, we’ll map this to OPENAI_API_KEY in app.py
API_KEY=sk-REPLACE_ME
# Deterministic local runs & tests without external calls
MOCK_MODE=1
PORT=8000
* client.py (httpx example)
# client.py
import httpx
payload = {
"text": "The app crashes on startup after the latest update. Please fix ASAP."
}
res = httpx.post("http://localhost:8000/classify", json=payload, timeout=30)
print(res.status_code, res.json())
* curl
curl -X POST http://localhost:8000/classify \
-H "Content-Type: application/json" \
-d '{"text":"Billing page shows wrong total; need a refund."}'
* Dockerfile (non-root)
# Dockerfile
FROM python:3.12-slim
ENV PIP_DISABLE_PIP_VERSION_CHECK=1 \
PYTHONDONTWRITEBYTECODE=1 \
PYTHONUNBUFFERED=1
# Create non-root user
RUN adduser --disabled-password --gecos "" appuser
WORKDIR /app
# System deps if needed later (kept minimal here)
RUN apt-get update && apt-get install -y --no-install-recommends \
ca-certificates && rm -rf /var/lib/apt/lists/*
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
USER appuser
EXPOSE 8000
CMD ["python", "app.py"]
CORS note: If you’re calling from a browser frontend, add flask-cors or set a proxy. Keep tokens server-side.
7) Verify output equivalence (deterministic test)
# tests/test_app.py
import os
os.environ["MOCK_MODE"] = "1" # deterministic, no external calls
import json
from app import app
def test_classify_basic():
client = app.test_client()
payload = {"text": "The settings screen throws an error on save. Crash log attached."}
resp = client.post("/classify", json=payload)
assert resp.status_code == 200
data = resp.get_json()
assert data["label"] in {"bug","feature","billing"}
words = len(data["summary"].split())
assert 5 <= words <= 80
Run tests:
pytest -q
8) Costs, rate limits, and guardrails
- Rough cost math (example): If your chosen model is ~$0.15 / 1K input tokens and ~$0.60 / 1K output tokens, and each request uses ~600 input + ~80 output tokens: per-request ≈ 0.6 * 0.15/1000 + 0.08 * 0.60/1000 ≈ $0.000135 + $0.000048 ≈ $0.000183 ≈ $0.18 per 1,000 requests. Substitute your model’s actual prices.
- Rate limits: Wrap calls with Tenacity backoff (already included). For bursts, queue work and process with workers.
- Dead-letter idea: If parsing fails after retries, push the payload to a “DLQ” (Redis list or SQS) and notify Slack for human review.
9) Privacy & compliance (PII)
- Redact before send: Strip emails, phone numbers, and IDs when possible.
- Logging policy: Log request IDs, not raw text. Store summaries + labels; keep originals only if required.
- Retention: Set a TTL for raw inputs. Encrypt at rest. Document your policy.
10) Pricing & ROI framing (map to outcomes)
- Starter ($29/mo): 10k requests, shared model, best-effort support.
- Team ($99/mo): 50k requests, alerting, Slack webhook, 99.9% uptime.
- Enterprise (custom): SSO, on-prem or VPC, custom label sets, signed DPA.
Pitch with time saved (e.g., 45 mins/day per agent) and reduced response latency.
11) Deployment: boring but safe
- Config: Use .env in dev; secrets in a manager in prod.
- Health check: /health returns {ok:true}.
- Rolling deploys: Blue/green or canary; keep a previous image tag ready.
- Rollback: If error rate > threshold, roll back automatically.
- Observability: Access logs + latency + error rate dashboards. (Add gunicorn for prod.)
12) Integration ideas (next steps)
- Helpdesk: Zendesk/Intercom webhooks → /classify → add label, suggested reply.
- Slack triage: Send summaries to a channel with buttons: Assign / Escalate.
- Notion/Sheets: Append {summary,label,link} rows for leaders.
- Queues: Celery workers with Redis for high volume.
- Playwright web automation: Auto-pull tickets from legacy portals.
- CLI: Typer CLI for batch files.
- Scaling: If a step is CPU-hot, try C++ with pybind11.
- Documents: If you go into PDF data extraction later, add OCR with Tesseract and parsing via PyMuPDF (fitz).
- Logging: Structured logs (e.g., loguru) when you expand.
13) FAQ
Q1. Why Flask over FastAPI? It’s a simple, familiar FastAPI alternative with tiny setup and plenty of docs.
Q2. Which model should I pick? Start with a small, inexpensive chat model. Swap via litellm without refactors.
Q3. How do I change the label set? Update ALLOWED and the system prompt. Add tests to lock behavior.
Q4. What about multilingual mail? Add “Detect language → summarize in English → keep label rules”. Works fine.
Q5. Can I keep it offline? For air-gapped requirements, use a local model endpoint compatible with litellm.
Q6. Is the test really offline? Yes — MOCK_MODE=1 avoids network calls and uses heuristics.
Q7. Can this power invoice processing automation? Yes; replace input sources and add a doc pipeline. The endpoint shape stays useful.
Q8. Any gotchas with Docker? Expose 8000, pass env vars, and front it with Nginx/Gunicorn for production.
14) Closing If this saved you time, star the repo, comment with your use case, and share metrics (time saved, resolution rate). I’ll add the best playbooks in a follow-up.
APPENDIX A) README (drop into your GitHub)
# Tiny Flask Classifier — Python Automation Tool
A minimal **Flask** service that returns `{label, summary}` for any text.
- Provider-agnostic via **litellm**
- **Pydantic v2** schemas, **Tenacity** retries
- **MOCK_MODE** for offline testing
- Docker-ready
## Quickstart
```bash
python -m venv .venv && source .venv/bin/activate # (Windows: .venv\Scripts\activate)
pip install -r requirements.txt
cp .env.example .env # set MODEL/API_KEY; keep MOCK_MODE=1 for local
python app.py
Call It
curl -X POST http://localhost:8000/classify \
-H "Content-Type: application/json" \
-d '{"text":"Payment failed and invoice shows wrong tax."}'
Tests
pytest -q
Docker
pytest -q
Env
- MODEL — model name supported by litellm
- API_KEY — provider key (mapped to OPENAI_API_KEY if present)
- MOCK_MODE — 1 for deterministic heuristics; 0 to call the model
- PORT — default 8000
Production Notes
- Use Gunicorn behind Nginx or a managed platform.
- Add rate limits, request auth, and structured logging.
- For browser apps, enable CORS at the gateway/proxy.
Let’s Connect + Fuel the Next Deep Dive ☕️ I build at the intersection of AI, code, and automation — and I share hands-on guides from LLMs to real-world apps. If this helped, stay in touch (and consider tipping a coffee to keep these coming):
- 📧 Email: [email protected]
- 💼 LinkedIn: Tarun Singh
- 💻 GitHub: github.com/krtarunsingh
- ☕ Buy Me a Coffee: https://buymeacoffee.com/krtarunsingh
👉 If you found value here, clap, share, and leave a comment — it helps more devs discover practical guides like this.
Build an AI PDF Search Engine in a Weekend (Python, FAISS, RAG — Full Code) Turn messy folders of PDFs into a blazing-fast, AI-assisted knowledge base you can actually talk to. pub.towardsai.net
The Next AI Boom: What Comes After AI Agents and Agentic AI? Artificial Intelligence is no longer science fiction. It’s a living, breathing force that’s already transforming how we… medium.com
AI-Powered OCR with Phi-3-Vision-128K: The Future of Document Processing In the fast-evolving world of artificial intelligence, multimodal models are setting new standards for integrating… ai.gopubby.com
RAG Frameworks Explored: LlamaIndex vs. LangChain for Next-Gen LLMs The Rise of Large Language Models (LLMs) ai.gopubby.com
Mastering RAG Chunking Techniques for Enhanced Document Processing Dividing large documents into smaller parts is a crucial yet intricate task that significantly impacts the performance… ai.gopubby.com
Read the full article here: https://pub.towardsai.net/from-weekend-hack-to-side-income-python-automation-with-flask-2e751b46539a