About This Code Showcase
This curated code walkthrough demonstrates how the Marketing Agency dashboard aggregates content drafts, lead data, publishing pipelines, and weekly reports into a single 4-tab interface for solo consultancy firms.
The system reads from local file repositories, connects to a lead management API, and queries git history to surface recent activity -- all served through a FastAPI backend with Jinja2 templates.
Project File Structure
marketing-agency/
app.py
config.py
readers.py
requirements.txt
start.bat
user-guide.md
static/
style.css
logo.png
templates/
dashboard.html
Core: FastAPI Dashboard Server
The main application mounts static files, configures Jinja2 templates, and serves a single dashboard endpoint that aggregates data from all four tabs -- Content, Leads, Pipeline, and Weekly Report.
"""
app.py -- Marketing Agency Dashboard
FastAPI application with 4-tab dashboard: Leads, Content, Pipeline, Weekly Report.
"""
from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from config import DASHBOARD_PORT
from readers import (
get_linkedin_drafts, get_posted_log, get_recent_blog_posts,
get_blog_drafts, get_publish_queue, get_publish_log,
get_case_study_stats, get_next_week_plan,
get_weekly_reports, get_leads_from_api, get_lead_stats_from_api,
get_recent_commits,
)
app = FastAPI(title="Marketing Agency Dashboard")
app.mount("/static", StaticFiles(directory="static"), name="static")
templates = Jinja2Templates(directory="templates")
@app.get("/", response_class=HTMLResponse)
async def dashboard(request: Request, tab: str = "content"):
linkedin_drafts = get_linkedin_drafts()
posted_log = get_posted_log()
recent_blogs = get_recent_blog_posts()
blog_drafts = get_blog_drafts()
publish_queue = get_publish_queue()
publish_log = get_publish_log()
case_study_stats = get_case_study_stats()
next_week_plan = get_next_week_plan()
weekly_reports = get_weekly_reports()
leads = get_leads_from_api()
lead_stats = get_lead_stats_from_api()
recent_commits = get_recent_commits()
posted_files = set()
for line in posted_log:
parts = line.split("|")
if len(parts) >= 2:
posted_files.add(parts[1].strip())
pending_posts = [d for d in linkedin_drafts
if d["filename"] not in posted_files]
return templates.TemplateResponse("dashboard.html", {
"request": request,
"tab": tab,
"linkedin_drafts": linkedin_drafts,
"pending_posts": pending_posts,
"posted_count": len(posted_files),
"recent_blogs": recent_blogs,
"blog_drafts": blog_drafts,
"publish_queue": publish_queue,
"publish_log": publish_log,
"case_study_stats": case_study_stats,
"next_week_plan": next_week_plan,
"weekly_reports": weekly_reports,
"leads": leads,
"lead_stats": lead_stats,
"recent_commits": recent_commits,
})
@app.get("/health")
async def health():
return {"status": "ok", "app": "Marketing Agency Dashboard"}
if __name__ == "__main__":
import uvicorn
uvicorn.run("app:app", host="0.0.0.0", port=DASHBOARD_PORT, reload=True)
Configuration: Centralized Path Management
All file paths and configuration constants live in a single module, making it easy to adapt the system for any consultancy firm's directory structure.
"""
config.py -- Marketing Agency Dashboard Configuration
All paths and constants in one place.
"""
import os
CONTENT_REPO = r"C:\Users\Lenovo\pau-analytics-1.0"
BLOG_DIR = os.path.join(CONTENT_REPO, "blog")
BLOG_DRAFTS_DIR = os.path.join(BLOG_DIR, "drafts")
LINKEDIN_DRAFTS_DIR = os.path.join(BLOG_DIR, "linkedin-drafts")
CASE_STUDY_DIR = os.path.join(CONTENT_REPO, "case-study")
PUBLISH_QUEUE = os.path.join(CONTENT_REPO, "publish-queue.md")
PUBLISH_LOG = os.path.join(CONTENT_REPO, "publish-log.txt")
NEXT_WEEK_PLAN = os.path.join(CONTENT_REPO, "next-week-plan.md")
WEEKLY_REPORTS_DIR = os.path.join(CONTENT_REPO, "reports", "weekly")
POSTED_LOG = os.path.join(LINKEDIN_DRAFTS_DIR, "posted-log.txt")
BLOG_INDEX = os.path.join(BLOG_DIR, "index.html")
LEAD_MANAGER_URL = os.getenv("LEAD_MANAGER_URL", "http://localhost:8000")
DASHBOARD_PORT = 8100
Data Readers: Content & Lead Aggregation
The readers module provides specialized functions that scan local directories for content drafts and query the lead management API. Each function returns structured data ready for the Jinja2 template.
def get_linkedin_drafts() -> list[dict]:
"""Scan the LinkedIn drafts directory for .txt files,
classify each by post type, and return metadata + preview."""
drafts = []
if not os.path.isdir(LINKEDIN_DRAFTS_DIR):
return drafts
for f in sorted(glob.glob(
os.path.join(LINKEDIN_DRAFTS_DIR, "*.txt")), reverse=True):
basename = os.path.basename(f)
if basename == "posted-log.txt":
continue
mod_time = datetime.fromtimestamp(os.path.getmtime(f))
content = read_file_safe(f)
if basename.startswith("insight-"):
post_type = "Insight Post"
else:
post_type = "Blog Teaser"
drafts.append({
"filename": basename,
"post_type": post_type,
"date": mod_time.strftime("%Y-%m-%d"),
"preview": content[:200].strip() if content else "(empty)",
"full_content": content.strip(),
"word_count": len(content.split()) if content else 0,
})
return drafts
def get_leads_from_api() -> list[dict]:
"""Fetch all leads from the Web Chat Lead Manager API.
Returns an empty list on timeout or connection error,
keeping the dashboard functional even when the API is down."""
try:
resp = httpx.get(
f"{LEAD_MANAGER_URL}/api/leads", timeout=2
)
if resp.status_code == 200:
return resp.json()
except Exception:
pass
return []
def get_lead_stats_from_api() -> dict:
"""Fetch aggregated lead statistics from the API.
Returns safe defaults so the dashboard renders correctly
even when the lead manager service is unavailable."""
try:
resp = httpx.get(
f"{LEAD_MANAGER_URL}/api/stats", timeout=2
)
if resp.status_code == 200:
return resp.json()
except Exception:
pass
return {
"total": 0,
"this_week": 0,
"needs_followup": 0,
"ch_a_week": 0,
"ch_b_week": 0,
"ch_w_week": 0,
"blog_performance": [],
}