AI Engineer · AI Systems Architect · Detroit, MI
A Python-first system for turning ideas and AI-assisted workflows into structured, reusable software. AI does the parts that need reasoning; everything repeatable moves into Python that runs the same way every time.
create and improve one component each, inside their own folder
connect proven components into larger workflows
validated systems are kept and reused, never rebuilt from zero
idea ──▶ AI reasons ──▶ Python builds it ──▶ gates check it ──▶ it runs on a schedule
│ │
└──────── kept, versioned, reused ◀──────┘Validation, reproducibility, cost control and failure recovery are built in, not added after.
Python crawls, a local Llama (Ollama) reads, Python checks. The model only turns clean text into structured JSON; design is my own components, never a model. Tested on a real page: valid JSON in under 10 seconds at $0, and the validator caught the facts the small model got wrong.
sitemap · ~20 pages · Playwright + Trafilatura
Ollama · JSON schema only · temperature 0
my components · no model
chunks · local embeddings · FAISS
no source string, no fact → export
URL
│
▼
① PYTHON CRAWLER
│
sitemap + navigation
max ~20 useful pages
│
┌────────┴────────┐
▼ ▼
PLAYWRIGHT TRAFILATURA
DOM/rendering clean text
screenshots content
│ │
└────────┬────────┘
▼
② FACT ENGINE
│
LOCAL LLAMA
│
STRUCTURED JSON ONLY
│
┌───────────────┼────────────────┐
▼ ▼ ▼
SERVICES FAQ PRICING
CONTACTS LOCATIONS PRODUCTS
│
└───────────────┬────────────────┘
▼
COMPANY_MODEL.JSON
│
┌─────────┴─────────┐
▼ ▼
③ CAI DESIGN ENGINE ④ RAG ENGINE
│ │
YOUR components chunk facts
YOUR spacing embeddings
YOUR palettes FAISS
YOUR typography metadata
│ │
▼ ▼
WEBSITE BUILD CHATBOT
│ │
└─────────┬─────────┘
▼
⑤ VALIDATOR
│
▼
EXPORTStatus: testing. The fact step has run on a live page; the rest is being wired.
One command lays out a whole project: a menu that finds every builder, a collector that feeds them, numbered stages inside each one. Every function is named so it reads like a sentence, checks what comes in, and reports what it did. This is how I keep code readable after years of changes, and how I find a failure in one look.
loop menu · finds every run.py
new · append · filter → _df_collect.csv
each builder's backbone
_0 get · _1 clean · _2 features · _3 model · _4 eval · _5 utils
raw → processed → exports
_1_build_audiofolders_create_GET_paths
↓
_2_proc_audio_split_GET_chunksdf
↓
_3_ext_audiofreq_detect_GET_freqdf
↓
_4_ana_audioquality_score_GET_statsdfThe name tells you the stage, what it does and what it gives back, before you open the file.
-----######----- FUNCTION SUMMARY -----######-----
FUNC : {function_name}
STAGE : {stage}
INPUT : {input_count}
OUTPUT : {output_count}
ERRORS : {n_errors}
TIME : {runtime}s
STATUS : SUCCESS / FAIL<project_name>/
├── auto.py ← loop menu; discovers run.py files; [C] pinned for collector
├── collector.py ← smart collector with interactive menu; outputs _df_collect.csv
├── README.md
├── requirements.txt
├── .gitignore
│
├── builders_<domain>/
│ ├── builder_1/
│ │ ├── run.py ← builder backbone (called by auto.py)
│ │ ├── src_1/ ← all stage files live here
│ │ │ ├── _0_fn_get_data.py
│ │ │ ├── _1_fn_clean_data.py
│ │ │ ├── _2_fn_features.py
│ │ │ ├── _3_fn_model.py
│ │ │ ├── _4_fn_eval.py
│ │ │ └── _5_fn_utils.py
│ │ └── data/
│ │ ├── raw/
│ │ ├── processed/
│ │ └── exports/
│ ├── builder_2/ (run.py + src_2/ + data/)
│ └── builder_3/ (run.py + src_3/ + data/)
│
└── config/
└── config.pyimport os
import sys
import subprocess
ROOT = os.path.dirname(os.path.abspath(__file__))
CSV_PATH = os.path.join(ROOT, "_df_collect.csv")
def clear():
os.system("cls" if os.name == "nt" else "clear")
def df_status():
if os.path.exists(CSV_PATH):
try:
import pandas as pd
df = pd.read_csv(CSV_PATH)
return f"df available {len(df)} rows"
except Exception:
return "df exists (unreadable)"
return "no df"
def find_runners():
runners = []
for entry in sorted(os.listdir(ROOT)):
if entry.startswith("builders_"):
domain_path = os.path.join(ROOT, entry)
for builder in sorted(os.listdir(domain_path)):
run_py = os.path.join(domain_path, builder, "run.py")
if os.path.exists(run_py):
runners.append((builder, run_py))
return runners
def show_menu(runners):
print("\n" + "=" * 44)
print(" MAIN MENU")
print("=" * 44)
for i, (name, _) in enumerate(runners, 1):
print(f" [{i}] {name}")
print()
print(f" [C] Collector ({df_status()})")
print(" [Q] Quit")
print("=" * 44)
def main():
while True:
clear()
runners = find_runners()
show_menu(runners)
choice = input("\n Select: ").strip().lower()
if choice == "q":
clear()
break
elif choice == "c":
sys.path.insert(0, ROOT)
import collector
import importlib
importlib.reload(collector)
collector.menu()
elif choice.isdigit():
idx = int(choice) - 1
if 0 <= idx < len(runners):
name, run_path = runners[idx]
clear()
print(f"\n Running {name}...\n")
subprocess.run([sys.executable, run_path])
input("\n Press Enter to return to menu...")
else:
input(" Invalid selection — Press Enter...")
else:
input(" Invalid selection — Press Enter...")
if __name__ == "__main__":
main()-----######----- CORE FUNCTION -----######-----
def _2_proc_audio_split_GET_chunksdf(df, output_dir, ...):
import os
from tqdm import tqdm
import time
function_name = "_2_proc_audio_split_GET_chunksdf"
stage = "PROC"
t0 = time.time()
# 1. VALIDATION
assert 'Path' in df.columns, "Missing 'Path' column"
# 2. SETUP
results = []
n_errors = 0
# 3. TQDM LOOP
for path in tqdm(df['Path'], desc='[PROC] Splitting Audio'):
try:
# 4. CORE LOGIC
# (your processing here)
# 5. STORE
results.append(...)
except Exception:
n_errors += 1
# 6. OUTPUT
df_chunks = ...
# 7. SUMMARY PRINT
runtime = round(time.time() - t0, 2)
print(f"""
-----######----- FUNCTION SUMMARY -----######----- FUNC : {function_name} STAGE : {stage} INPUT : {len(df)} OUTPUT : {len(df_chunks)} ERRORS : {n_errors} TIME : {runtime}s STATUS : SUCCESS
""")
return df_chunks# /scaffold — Create a new app project with standard structure
Scaffold a new project using the user's standard layout.
## Usage
```
/scaffold <project_name> <domain>
```
Examples:
- `/scaffold music_code_structure music`
- `/scaffold finance_pipeline finance`
## Directory structure
```
<project_name>/
├── auto.py ← loop menu; discovers run.py files; [C] pinned for collector
├── collector.py ← smart collector with interactive menu; outputs _df_collect.csv
├── README.md
├── requirements.txt
├── .gitignore
│
├── builders_<domain>/
│ ├── builder_1/
│ │ ├── run.py ← builder backbone (called by auto.py)
│ │ ├── src_1/ ← all stage files live here
│ │ │ ├── _0_fn_get_data.py
│ │ │ ├── _1_fn_clean_data.py
│ │ │ ├── _2_fn_features.py
│ │ │ ├── _3_fn_model.py
│ │ │ ├── _4_fn_eval.py
│ │ │ └── _5_fn_utils.py
│ │ └── data/
│ │ ├── raw/
│ │ ├── processed/
│ │ └── exports/
│ ├── builder_2/ (run.py + src_2/ + data/)
│ └── builder_3/ (run.py + src_3/ + data/)
│
└── config/
└── config.py
```
Data folders get a `.gitkeep`. Stage files inside `src_N/` are empty skeletons (`pass`).
## auto.py — full implementation
```python
import os
import sys
import subprocess
ROOT = os.path.dirname(os.path.abspath(__file__))
CSV_PATH = os.path.join(ROOT, "_df_collect.csv")
def clear():
os.system("cls" if os.name == "nt" else "clear")
def df_status():
if os.path.exists(CSV_PATH):
try:
import pandas as pd
df = pd.read_csv(CSV_PATH)
return f"df available {len(df)} rows"
except Exception:
return "df exists (unreadable)"
return "no df"
def find_runners():
runners = []
for entry in sorted(os.listdir(ROOT)):
if entry.startswith("builders_"):
domain_path = os.path.join(ROOT, entry)
for builder in sorted(os.listdir(domain_path)):
run_py = os.path.join(domain_path, builder, "run.py")
if os.path.exists(run_py):
runners.append((builder, run_py))
return runners
def show_menu(runners):
print("\n" + "=" * 44)
print(" MAIN MENU")
print("=" * 44)
for i, (name, _) in enumerate(runners, 1):
print(f" [{i}] {name}")
print()
print(f" [C] Collector ({df_status()})")
print(" [Q] Quit")
print("=" * 44)
def main():
while True:
clear()
runners = find_runners()
show_menu(runners)
choice = input("\n Select: ").strip().lower()
if choice == "q":
clear()
break
elif choice == "c":
sys.path.insert(0, ROOT)
import collector
import importlib
importlib.reload(collector)
collector.menu()
elif choice.isdigit():
idx = int(choice) - 1
if 0 <= idx < len(runners):
name, run_path = runners[idx]
clear()
print(f"\n Running {name}...\n")
subprocess.run([sys.executable, run_path])
input("\n Press Enter to return to menu...")
else:
input(" Invalid selection — Press Enter...")
else:
input(" Invalid selection — Press Enter...")
if __name__ == "__main__":
main()
```
## collector.py — full implementation
Copy the full collector.py from `music_code_structure/collector.py` as the canonical template.
It implements: `_new`, `_append`, `_replace`, `_view`, `_filter`, `_clear`, `menu()`, `main()`.
Menu options:
- **[N] New** — fresh collection, replaces df
- **[A] Append** — adds rows, dedupes by file_path
- **[R] Replace** — preview current df then swap with new input
- **[V] View** — shape, columns, extension breakdown with bar, total size MB, head(5)
- **[F] Filter** — filter by extension(s), optionally save filtered result
- **[X] Clear** — delete _df_collect.csv with confirmation
- **[B] Back** — return to auto.py menu
Status line always shows: row count + top extensions with counts.
## Other file rules
### run.py (each builder)
- Empty skeleton: `pass`
### config/config.py
- Empty skeleton: `pass`
### requirements.txt
```
pandas
```
### .gitignore
```
__pycache__/
*.pyc
_df_collect.csv
data/raw/*
data/processed/*
data/exports/*
!**/.gitkeep
```
## After scaffolding
Print a short tree confirming the created structure.# 🧱 FUNCTION SYSTEM — FULL STRUCTURE (SCALABLE)
---
## 1. FILE STRUCTURE (Stage-Based)
### Rule: **One stage per file**
src/
_1_build_audio.py _2_proc_audio.py _3_ext_audio.py _4_ana_audio.py _5_util_audio.py
- `_1` → build (setup, folders, paths)
- `_2` → process (transformations)
- `_3` → extract (features, metadata)
- `_4` → analyze (scoring, insights)
- `_5` → utilities (helpers)
---
## 2. FUNCTION NAMING (LOCK THIS)
### Format:
_#stage##_type_keyword##_action_GET_output
### Examples:
_1_build_audiofolders_create_GET_paths _2_proc_audio_split_GET_chunksdf _3_ext_audiofreq_detect_GET_freqdf _4_ana_audioquality_score_GET_statsdf
---
## 3. PIPELINE FLOW (Readable Story)
_1_build → _2_proc → _3_ext → _4_ana
### Example:
_1_build_audiofolders_create_GET_paths ↓ _2_proc_audio_split_GET_chunksdf ↓ _3_ext_audiofreq_detect_GET_freqdf ↓ _4_ana_audioquality_score_GET_statsdf
---
## 4. CORE FUNCTION TEMPLATE
-----######----- CORE FUNCTION -----######-----
def _2_proc_audio_split_GET_chunksdf(df, output_dir, ...):
import os
from tqdm import tqdm
import time
function_name = "_2_proc_audio_split_GET_chunksdf"
stage = "PROC"
t0 = time.time()
# 1. VALIDATION
assert 'Path' in df.columns, "Missing 'Path' column"
# 2. SETUP
results = []
n_errors = 0
# 3. TQDM LOOP
for path in tqdm(df['Path'], desc='[PROC] Splitting Audio'):
try:
# 4. CORE LOGIC
# (your processing here)
# 5. STORE
results.append(...)
except Exception:
n_errors += 1
# 6. OUTPUT
df_chunks = ...
# 7. SUMMARY PRINT
runtime = round(time.time() - t0, 2)
print(f"""
-----######----- FUNCTION SUMMARY -----######----- FUNC : {function_name} STAGE : {stage} INPUT : {len(df)} OUTPUT : {len(df_chunks)} ERRORS : {n_errors} TIME : {runtime}s STATUS : SUCCESS
""")
return df_chunks
---
## 5. STANDARD PRINT FORMAT
### Always use:
-----######----- FUNCTION SUMMARY -----######----- FUNC : {function_name} STAGE : {stage} INPUT : {input_count} OUTPUT : {output_count} ERRORS : {n_errors} TIME : {runtime}s STATUS : SUCCESS / FAIL
---
## 6. RUNNING PIPELINE
#!#!#!#!#! RUNNING STATEMENTS #!#!#!#!#!
df_1 = _1_build_audiofolders_create_GET_paths(...) df_2 = _2_proc_audio_split_GET_chunksdf(df_1, ...) df_3 = _3_ext_audiofreq_detect_GET_freqdf(df_2, ...) df_4 = _4_ana_audioquality_score_GET_statsdf(df_3)
---
## 7. HARD RULES
- One function = one responsibility
- One file = one stage
- Always return usable output (df or structured object)
- Always include tqdm for loops
- Always include summary print
- Never mix stages in one function
- Naming must reflect:
- stage
- action
- output
---
## 8. MENTAL MODEL
- File = **Stage**
- Function = **Action**
- Pipeline = **Story**
- Print = **Report**
---
## 9. QUICK CHECKLIST BEFORE CODING
- [ ] Input clearly defined
- [ ] Output clearly defined
- [ ] Stage assigned (_1, _2, _3...)
- [ ] Function name follows format
- [ ] tqdm included if looping
- [ ] Summary print included
- [ ] Returns structured output🧪 3. VALIDATION LAYER (INPUT + OUTPUT CHECKS)
🔹 Why
* Prevent silent errors → critical for pipelines
🔹 Add pattern:
def _validate_df_has_cols(df, cols):
for c in cols:
if c not in df.columns:
raise ValueError(f"Missing column: {c}")
🔹 Also validate output:
assert len(df_chunks) > 0, "No chunks created"
🔹 Benefit
* Fail fast instead of corrupting pipeline
⚡ BONUS (HIGH IMPACT) 🔹 6. ERROR BUCKET SYSTEM Instead of losing failures: errors = [] try: ... except Exception as e: errors.append((path, str(e))) Return: return df_chunks, errorsWritten as a Claude Code command, so every project an agent scaffolds starts in the same shape.
/struct lays out any data project so it runs on any machine without one edit. Its web twin keeps every law and swaps the parts: builders become layers, the dataframe becomes the site, and a landing page that asks for input in steps ships with its own run test.
layer 1 ──Next──▶ layer 2 ──Next──▶ layer 3 ──Send──▶ server checks it all again ──▶ stored
│ │ │ │
└── page checks ──┴── page checks ──┘ steps.jsonl ──▶ funnel per layer ◀──┘# /struct — Create a new app project with standard structure
This is the general-purpose project structure used across all domains (music, finance, etc.).
## ⚠️ PORTABILITY RULE — ENFORCE ALWAYS
**This app must work on any machine, in any folder, without modification.**
- **ZERO hardcoded external paths** anywhere in the project — not in functions, not in run.py, not in config files
- All internal paths are computed at runtime using `os.path.dirname(os.path.abspath(__file__))` as the anchor
- `_df_collect.csv` lives at project root and is always located relative to `__file__`
- The **only** file that ever touches external/desktop paths is `export_2_desk.py` — that is its entire purpose
- If you ever see an absolute path like `/Users/...` or `C:\...` inside any file except `export_2_desk.py`, it is a bug
## ROLE OF EACH ROOT FILE
| File | Role |
|------|------|
| `auto.py` | Loop menu — orchestrates builders and collector |
| `collector.py` | Builds `_df_collect.csv` — the df we work on |
| `export_2_desk.py` | **Only file with external/desktop paths** — highly customizable per project; copies builder exports to desktop with appropriate naming/format |
| `_df_collect.csv` | The working dataframe — input to all builders |
## Collector note
`collector.py` is the one file that may vary by domain. When the user invokes `/struct` for the first time on a new project, ask: **"Does this project need a custom collector, or use the standard file/folder collector?"**
- **Standard**: file/folder walker → `_df_collect.csv` (copy as-is from template)
- **Custom**: user will describe the data source (API, database, spreadsheet, etc.) — adapt `collector.py` for that source while keeping the same menu structure (`menu()`) and CSV output contract (`_df_collect.csv`)
## Usage
```
/struct <project_name> <domain>
```
Examples:
- `/struct music_code_structure music`
- `/struct finance_pipeline finance`
## Directory structure
```
<project_name>/
├── auto.py ← loop menu; discovers run.py files; [C] pinned for collector
├── collector.py ← smart collector; outputs _df_collect.csv
├── export_2_desk.py ← ONLY file with external paths; copies exports → Desktop
├── _df_collect.csv ← working df (gitignored; created by collector)
├── README.md
├── requirements.txt
├── .gitignore
│
├── builders_<domain>/
│ ├── builder_1/
│ │ ├── run.py ← builder backbone (called by auto.py)
│ │ ├── src_1/ ← stage files (named by /pystr convention)
│ │ │ ├── _1_build_<domain>.py
│ │ │ ├── _2_proc_<domain>.py
│ │ │ ├── _3_ext_<domain>.py
│ │ │ ├── _4_ana_<domain>.py
│ │ │ └── _5_util_<domain>.py
│ │ └── data/
│ │ ├── raw/
│ │ ├── processed/
│ │ └── exports/
│ ├── builder_2/ (run.py + src_2/ + data/)
│ └── builder_3/ (run.py + src_3/ + data/)
│
└── config/
├── config.py ← global app config
├── config_u1.py ← user-editable config for builder_1
├── config_u2.py ← user-editable config for builder_2
└── config_u3.py ← user-editable config for builder_3
```
Data folders get a `.gitkeep`. Stage files inside `src_N/` are empty skeletons (`pass`).
Each `config_uN.py` is the first import in its `run.py` — see /pystr Section 11 for the full pattern.
## auto.py — full implementation
```python
import os
import sys
import subprocess
ROOT = os.path.dirname(os.path.abspath(__file__))
CSV_PATH = os.path.join(ROOT, "_df_collect.csv")
def clear():
os.system("cls" if os.name == "nt" else "clear")
def df_status():
if os.path.exists(CSV_PATH):
try:
import pandas as pd
df = pd.read_csv(CSV_PATH)
return f"df available {len(df)} rows"
except Exception:
return "df exists (unreadable)"
return "no df"
def find_runners():
runners = []
for entry in sorted(os.listdir(ROOT)):
if entry.startswith("builders_"):
domain_path = os.path.join(ROOT, entry)
for builder in sorted(os.listdir(domain_path)):
run_py = os.path.join(domain_path, builder, "run.py")
if os.path.exists(run_py):
runners.append((builder, run_py))
return runners
def show_menu(runners):
print("\n" + "=" * 44)
print(" MAIN MENU")
print("=" * 44)
for i, (name, _) in enumerate(runners, 0):
print(f" [{i}] {name}")
print()
print(f" [C] Collector ({df_status()})")
print(" [Q] Quit")
print("=" * 44)
def main():
while True:
clear()
runners = find_runners()
show_menu(runners)
choice = input("\n Select: ").strip().lower()
if choice == "q":
clear()
break
elif choice == "c":
sys.path.insert(0, ROOT)
import collector
import importlib
importlib.reload(collector)
collector.menu()
elif choice.isdigit():
idx = int(choice)
if 0 <= idx < len(runners):
name, run_path = runners[idx]
clear()
print(f"\n Running {name}...\n")
subprocess.run([sys.executable, run_path])
input("\n Press Enter to return to menu...")
else:
input(" Invalid selection — Press Enter...")
else:
input(" Invalid selection — Press Enter...")
if __name__ == "__main__":
main()
```
## collector.py — full implementation
Copy the full collector.py from `music_code_structure/collector.py` as the canonical template.
It implements: `_new`, `_append`, `_replace`, `_view`, `_filter`, `_clear`, `menu()`, `main()`.
Menu options:
- **[N] New** — fresh collection, replaces df
- **[A] Append** — adds rows, dedupes by file_path
- **[R] Replace** — preview current df then swap with new input
- **[V] View** — shape, columns, extension breakdown with bar, total size MB, head(5)
- **[F] Filter** — filter by extension(s), optionally save filtered result
- **[X] Clear** — delete _df_collect.csv with confirmation
- **[B] Back** — return to auto.py menu
Status line always shows: row count + top extensions with counts.
## Other file rules
### run.py (each builder)
- Empty skeleton: `pass`
- First import must be its `config_uN.py` (e.g. `from config.config_u1 import *`)
### config/config.py
- Empty skeleton: `pass`
### config/config_uN.py (one per builder)
- Empty skeleton with comment: `# USER CONFIG — builder_N`
- Contains ONLY functional attributes: numbers, thresholds, formats, flags, extension lists, seeds
- NEVER contains paths — paths are always computed at runtime via `__file__`
### export_2_desk.py
- The one and only file allowed to use desktop/external paths (`os.path.expanduser("~/Desktop")` etc.)
- Reads from builder `data/exports/` folders (project-relative)
- Copies/exports to Desktop with appropriate naming
- Highly customizable per project: zip, rename, format conversion, selective export, etc.
- Skeleton only at scaffold time — user/AI fills it in based on project needs
### requirements.txt
```
pandas
```
### .gitignore
```
__pycache__/
*.pyc
_df_collect.csv
data/raw/*
data/processed/*
data/exports/*
!**/.gitkeep
```
## After creating
Print a short tree confirming the created structure.<project_name>/
├── auto.py ← loop menu; discovers run.py files; [C] pinned for collector
├── collector.py ← smart collector; outputs _df_collect.csv
├── export_2_desk.py ← ONLY file with external paths; copies exports → Desktop
├── _df_collect.csv ← working df (gitignored; created by collector)
├── README.md
├── requirements.txt
├── .gitignore
│
├── builders_<domain>/
│ ├── builder_1/
│ │ ├── run.py ← builder backbone (called by auto.py)
│ │ ├── src_1/ ← stage files (named by /pystr convention)
│ │ │ ├── _1_build_<domain>.py
│ │ │ ├── _2_proc_<domain>.py
│ │ │ ├── _3_ext_<domain>.py
│ │ │ ├── _4_ana_<domain>.py
│ │ │ └── _5_util_<domain>.py
│ │ └── data/
│ │ ├── raw/
│ │ ├── processed/
│ │ └── exports/
│ ├── builder_2/ (run.py + src_2/ + data/)
│ └── builder_3/ (run.py + src_3/ + data/)
│
└── config/
├── config.py ← global app config
├── config_u1.py ← user-editable config for builder_1
├── config_u2.py ← user-editable config for builder_2
└── config_u3.py ← user-editable config for builder_3<site_name>/
├── auto.py ← loop menu; lists layers; P · K · T · A · E · Q
├── run.py ← build → preview · --check · --test · --analyze
├── export_2_desk.py ← ONLY file with external paths
├── README.md
├── requirements.txt ← stdlib; playwright optional (browser test)
├── .gitignore
│
├── layers_<domain>/ ← one folder per input layer, shown one at a time
│ ├── layer_1_<step>/
│ │ ├── layer.html ← the fields, markup only (no form, no buttons)
│ │ ├── schema.json ← title + fields: name · type · label · required · options
│ │ └── layer.css ← optional, this layer only
│ ├── layer_2_<step>/
│ └── layer_3_<step>/
│
├── src_web/ ← stage files, /pystr names, contract fn(site, paths, cfg) → site
│ ├── _1_build_web.py ← find layers in number order, load parts
│ ├── _2_proc_web.py ← markup ⇄ schema must agree · unique names · known types · fail fast
│ ├── _3_ext_web.py ← stitch shell + layers → dist/ (index · css · js · schema.json)
│ ├── _4_ana_web.py ← submissions → csv · steps → funnel per layer
│ ├── _5_util_web.py ← print · files · validate_payload (the only cross-import)
│ └── shell/ ← index.html · site.css · site.js (the stepper)
│
├── tests/
│ ├── test_site.py ← the run test: build · portability · worst-case submits · funnel
│ └── test_browser.py ← optional: headless Chrome walks every layer at phone width
│
├── config/
│ ├── config.py ← global: name · port · limits · field types (no paths)
│ └── config_u1.py ← user-editable: the words on the page (no paths)
│
├── data/
│ ├── raw/ ← submissions.jsonl · steps.jsonl (append only)
│ ├── processed/ ← submissions.csv
│ └── exports/ ← funnel.csv
└── dist/ ← built site (gitignored)"""The site's front door. Python only, stdlib only.
python run.py build dist/ and preview it at 127.0.0.1:PORT (POST /submit · POST /step)
python run.py --check build only: exit 0 when every layer is sound, 1 when not
python run.py --test the run test: every check in tests/ (schema · build · portability · submit flow)
python run.py --analyze submissions → data/processed/submissions.csv · funnel → data/exports/funnel.csv"""
import json
import os
import sys
from datetime import datetime, timezone
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
from config import config, config_u1 # USER CONFIG first, always
_HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, _HERE)
from src_web._1_build_web import build # noqa: E402
from src_web._2_proc_web import process # noqa: E402
from src_web._3_ext_web import extract # noqa: E402
from src_web._4_ana_web import analyze # noqa: E402
from src_web._5_util_web import validate_payload # noqa: E402
CFG = {k: v for m in (config, config_u1) for k, v in vars(m).items() if k.isupper()}
PATHS = {
"layers": os.path.join(_HERE, "layers_web"),
"src": os.path.join(_HERE, "src_web"),
"dist": os.path.join(_HERE, "dist"),
"raw": os.path.join(_HERE, "data", "raw"),
"processed": os.path.join(_HERE, "data", "processed"),
"exports": os.path.join(_HERE, "data", "exports"),
}
def run(paths=PATHS, cfg=CFG):
print(f"\n ┌─ {cfg['SITE_NAME']} {'─' * (40 - len(cfg['SITE_NAME']))}┐")
site = {}
site = build(site, paths, cfg)
site = process(site, paths, cfg)
site = extract(site, paths, cfg)
print(" └──────────────────────────────────────────┘\n")
return site
def _append(path, row):
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "a", encoding="utf-8") as f:
f.write(json.dumps(row, ensure_ascii=False) + "\n")
def make_server(schema, paths=PATHS, cfg=CFG, port=None):
class Handler(SimpleHTTPRequestHandler):
def __init__(self, *a, **k):
super().__init__(*a, directory=paths["dist"], **k)
def log_message(self, *a):
pass
def _send(self, code, body):
data = json.dumps(body).encode()
self.send_response(code)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(data)))
self.end_headers()
self.wfile.write(data)
def do_POST(self):
n = int(self.headers.get("Content-Length") or 0)
if n > cfg["MAX_BODY_BYTES"]:
return self._send(413, {"ok": False, "errors": ["too big"]})
try:
body = json.loads(self.rfile.read(n) or b"null")
except ValueError:
return self._send(400, {"ok": False, "errors": ["not json"]})
now = datetime.now(timezone.utc).isoformat(timespec="seconds")
if self.path == "/submit":
errors = validate_payload(schema, body, cfg)
if errors:
return self._send(400, {"ok": False, "errors": errors})
_append(os.path.join(paths["raw"], "submissions.jsonl"), {"at": now, **body})
return self._send(200, {"ok": True})
if self.path == "/step" and isinstance(body, dict) and str(body.get("layer", "")).isdigit():
_append(os.path.join(paths["raw"], "steps.jsonl"),
{"at": now, "visit": str(body.get("visit", ""))[:16], "layer": int(body["layer"])})
return self._send(200, {"ok": True})
return self._send(404, {"ok": False, "errors": ["no such route"]})
return ThreadingHTTPServer(("127.0.0.1", cfg["PORT"] if port is None else port), Handler)
def main():
arg = sys.argv[1] if len(sys.argv) > 1 else ""
if arg == "--test":
import unittest
suite = unittest.defaultTestLoader.discover(os.path.join(_HERE, "tests"), top_level_dir=_HERE)
raise SystemExit(0 if unittest.TextTestRunner(verbosity=2).run(suite).wasSuccessful() else 1)
try:
site = run()
except ValueError as e:
print(f" ✗ {e}")
raise SystemExit(1)
if arg == "--check":
raise SystemExit(0)
if arg == "--analyze":
analyze(site, PATHS, CFG)
raise SystemExit(0)
srv = make_server(site["schema"])
print(f" preview http://127.0.0.1:{CFG['PORT']}/ (ctrl-c stops it)")
try:
srv.serve_forever()
except KeyboardInterrupt:
print("\n stopped")
if __name__ == "__main__":
main()"""_2 process — check every layer before anything is written. Markup and schema must agree. Fails fast."""
import re
import time
from src_web._5_util_web import hr, summary
def process(site, paths, cfg):
t0 = time.time()
hr("PROCESS")
errors, seen = [], set()
if not site["layers"]:
errors.append("no layers found")
for layer in site["layers"]:
name, sch = layer["name"], layer["schema"]
if not sch.get("title"):
errors.append(f"{name}: schema has no title")
if not sch.get("fields"):
errors.append(f"{name}: schema has no fields")
for f in sch.get("fields", []):
fn = f.get("name", "")
if not re.fullmatch(r"[a-z][a-z0-9_]*", fn):
errors.append(f"{name}: bad field name {fn!r}")
if fn in seen:
errors.append(f"{name}: field {fn} used twice")
seen.add(fn)
if f.get("type") not in cfg["FIELD_TYPES"]:
errors.append(f"{name}: {fn} has unknown type {f.get('type')!r}")
if f.get("type") == "choice" and not f.get("options"):
errors.append(f"{name}: choice {fn} has no options")
if f'name="{fn}"' not in layer["html"]:
errors.append(f"{name}: {fn} is in schema.json but not in layer.html")
for fn in re.findall(r'name="([^"]+)"', layer["html"]):
if fn not in {f.get("name") for f in sch.get("fields", [])}:
errors.append(f"{name}: {fn} is in layer.html but not in schema.json")
for e in errors:
print(f" ✗ {e}")
summary("_2_proc_webschema_check_GET_site", "PROC", len(site["layers"]), len(site["layers"]), len(errors), t0)
if errors:
raise ValueError(f"{len(errors)} layer error(s): fix them before building")
return site"""The run test. No network beyond 127.0.0.1, nothing written outside the project; test files are removed after."""
import json
import os
import re
import threading
import unittest
import urllib.error
import urllib.request
import run
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
TEST_RAW = os.path.join(ROOT, "data", "raw", "_test")
class Site(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.site = run.run()
cls.paths = {**run.PATHS, "raw": TEST_RAW}
cls.srv = run.make_server(cls.site["schema"], cls.paths, run.CFG, port=0)
cls.url = f"http://127.0.0.1:{cls.srv.server_address[1]}"
threading.Thread(target=cls.srv.serve_forever, daemon=True).start()
@classmethod
def tearDownClass(cls):
cls.srv.shutdown()
for f in ("submissions.jsonl", "steps.jsonl"):
p = os.path.join(TEST_RAW, f)
if os.path.exists(p):
os.remove(p)
if os.path.isdir(TEST_RAW):
os.rmdir(TEST_RAW)
def post(self, route, body):
req = urllib.request.Request(self.url + route, json.dumps(body).encode(), {"Content-Type": "application/json"})
try:
with urllib.request.urlopen(req) as r:
return r.status, json.loads(r.read())
except urllib.error.HTTPError as e:
return e.code, json.loads(e.read())
def good(self):
return {"need": "A chatbot", "need_words": "Ñandú · 名前 · 🚀 " + "x" * 1500,
"company": "O'Brien & Søn, Ltd.", "team_size": "12",
"name": "Maximiliano Fernández-Oyarzún de la Torre", "email": "a.very.long.name@sub.example-company.co",
"consent": True}
# ── the page ──────────────────────────────────────────────────────────────────────────────
def test_every_layer_in_order(self):
page = open(os.path.join(ROOT, "dist", "index.html"), encoding="utf-8").read()
found = [int(n) for n in re.findall(r'data-layer="(\d+)"', page)]
self.assertEqual(found, list(range(len(self.site["layers"]))))
self.assertEqual(page.count(" hidden>"), len(found) - 1 + 1) # all layers but the first, plus the done panel
def test_every_field_on_the_page(self):
page = open(os.path.join(ROOT, "dist", "index.html"), encoding="utf-8").read()
for layer in self.site["schema"]["layers"]:
for f in layer["fields"]:
self.assertIn(f'name="{f["name"]}"', page, f["name"])
def test_no_placeholder_left(self):
page = open(os.path.join(ROOT, "dist", "index.html"), encoding="utf-8").read()
self.assertNotIn("{{", page)
# ── portability: only export_2_desk.py may name a place outside the project ───────────────────
def test_no_outside_paths(self):
bad = re.compile("/" + "Users/|[A-Z]:" + r"\\\\" + "|expand" + "user|/" + "home/")
for d, _, files in os.walk(ROOT):
if "dist" in d.split(os.sep) or "__pycache__" in d:
continue
for f in files:
if f.endswith((".py", ".html", ".css", ".js", ".json")) and f != "export_2_desk.py":
text = open(os.path.join(d, f), encoding="utf-8").read()
self.assertIsNone(bad.search(text), os.path.join(d, f))
# ── the layers, broken on purpose ───────────────────────────────────────────────────────────
def test_schema_and_markup_must_agree(self):
broken = {"layers": [{"name": "layer_1_x", "html": '<input name="a">', "css": "",
"schema": {"title": "x", "fields": [{"name": "b", "type": "text"}]}}]}
with self.assertRaises(ValueError):
run.process(broken, run.PATHS, run.CFG)
def test_duplicate_field_refused(self):
lay = lambda n: {"name": n, "html": '<input name="a">', "css": "",
"schema": {"title": "x", "fields": [{"name": "a", "type": "text"}]}}
with self.assertRaises(ValueError):
run.process({"layers": [lay("layer_1_a"), lay("layer_2_b")]}, run.PATHS, run.CFG)
# ── the submit flow, worst cases first ─────────────────────────────────────────────────────
def test_empty_refused(self):
code, out = self.post("/submit", {})
self.assertEqual(code, 400)
self.assertTrue(any("required" in e for e in out["errors"]))
def test_bad_email_refused(self):
code, out = self.post("/submit", {**self.good(), "email": "not-an-email"})
self.assertEqual(code, 400)
self.assertIn("email: not an email", out["errors"])
def test_consent_must_be_ticked(self):
code, out = self.post("/submit", {**self.good(), "consent": "true"})
self.assertEqual(code, 400)
def test_unknown_field_refused(self):
code, out = self.post("/submit", {**self.good(), "admin": "yes"})
self.assertEqual(code, 400)
def test_choice_outside_options_refused(self):
code, _ = self.post("/submit", {**self.good(), "need": "Free money"})
self.assertEqual(code, 400)
def test_too_long_refused(self):
code, _ = self.post("/submit", {**self.good(), "need_words": "x" * 5000})
self.assertEqual(code, 400)
def test_not_json_refused(self):
req = urllib.request.Request(self.url + "/submit", b"{nope", {"Content-Type": "application/json"})
with self.assertRaises(urllib.error.HTTPError) as e:
urllib.request.urlopen(req)
self.assertEqual(e.exception.code, 400)
def test_worst_case_good_payload_stored(self):
code, out = self.post("/submit", self.good())
self.assertEqual((code, out), (200, {"ok": True}))
rows = [json.loads(l) for l in open(os.path.join(TEST_RAW, "submissions.jsonl"), encoding="utf-8")]
self.assertEqual(rows[-1]["name"], self.good()["name"])
def test_steps_feed_the_funnel(self):
for layer in range(3):
self.assertEqual(self.post("/step", {"visit": "t1", "layer": layer})[0], 200)
self.assertEqual(self.post("/step", {"visit": "t2", "layer": 0})[0], 200)
self.assertEqual(self.post("/step", {"visit": "t3", "layer": "drop table"})[0], 404)
paths = {**run.PATHS, "raw": TEST_RAW, "processed": TEST_RAW, "exports": TEST_RAW}
run.analyze(self.site, paths, run.CFG)
funnel = open(os.path.join(TEST_RAW, "funnel.csv"), encoding="utf-8").read().splitlines()
os.remove(os.path.join(TEST_RAW, "funnel.csv"))
os.remove(os.path.join(TEST_RAW, "submissions.csv"))
self.assertTrue(funnel[1].endswith(",2"), funnel) # two visits passed layer 1
self.assertTrue(funnel[3].endswith(",1"), funnel) # one visit passed layer 3
if __name__ == "__main__":
unittest.main()Stdlib only. Playwright is optional, for the browser walk.
Custom AI Models presents
One door for clients: sign in like anywhere else, then scroll a landing page that is also the menu. Every step up the ladder hands the client more control over their own AI.
Remote Control for Aggregated Integrations
Remote Control Source for Custom Aggregated Integrations
Source Remote Control for Custom Aggregated Integrations
Custom Training for Remote Aggregated Integrations Liaisons · trains people who find people
sign in ──▶ scroll ──▶ 01 ──▶ 02 ──▶ 03 ──▶ 04 ──▶ 05
scope clone +chat your AI agentic
─────────── more control for the client ──────────▶on top of the app you already have · custom, scoped as a project
scopedcustom adds · 6 months support · tool costs included · then from $15 with the URL · no surprise costs
$888a year of support · tool costs included · then subscription from $35
$2,499full AI development: an optimised web solution · a chatbot agent built from how customers really ask · your own editable AI + devops tool for the site · standard / silver / gold
tiersagentic remote control for your custom AI models
2027Status: beta. Prices and names are a working draft, not a published offer.
I’m also building CascForward, a Python-first system for turning ideas and AI-assisted workflows into structured, reusable software. CascForward: build it, automate it, preserve what works. The idea is simple. Use AI for the parts that require reasoning, then move repeatable work into reliable Python systems. Builders create and improve components, automations connect them into larger workflows, and validated systems can be preserved and reused instead of rebuilt every time.
CAI - HUB - BETA Custom AI Models presents =——— R-CAI Remote Control for Aggregated Integrations RCS-CAI Remote Control Source for Custom Aggregated Integrations SRC-CAI Source Remote Control for Custom Aggregated Integrations =——— CTRAIL Custom training for Remote Aggregated Integrations Liaisons (Recruits ppl to help you find ppl ) - trainings =——— Design Page . Click “R-CAI” goes in -> Whats inside designs wise : Log in / Sign up (this has to be very similar as other standard entry points ) Is basecally also a landing page , scroll down you will see 1) add/edit a web app to existing web app (custom side - scope a project ) 2) like the package 888 for cloning the website , custom adds will be there -> 6 months support added , tool cost included , after 6 months starts at 15 with URL - no surprise costs 3. Clone website and add chatbot - 2499 -> yearly support included , tool cost included , after a year subscription starts 35 4. SRCCAI - full AI development w source remote control your own custom AI model for you business a) optimized web solution b) included Chat bot agent -> dev from how customers interact c) your own AI editable + dev ops tool for your website (standard /silver/gold) 5. ARCCAI - AGENTIC REMOTE control for your Custom AI models - coming up 2027 Bersecally this is what we are selling .