import sqlite3, os, threading, webbrowser, json from datetime import datetime, timedelta from flask import Flask, render_template, request, redirect, url_for, jsonify, send_file, session from functools import wraps from functools import wraps from reportlab.lib.pagesizes import A4 from reportlab.lib import colors from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, Table, TableStyle, HRFlowable, Image as RLImage from reportlab.lib.units import cm BASE_DIR = os.path.dirname(os.path.abspath(__file__)) DB_PATH = os.path.join(BASE_DIR, "incidents.db") RPT_DIR = os.path.join(BASE_DIR, "reports") os.makedirs(RPT_DIR, exist_ok=True) app = Flask(__name__) app.secret_key = "connectgh_secret_2026_xK9mP2" # Login credentials — change these APP_USERNAME = "admin" APP_PASSWORD = "connectgh2026" def login_required(f): @wraps(f) def decorated(*args, **kwargs): if not session.get("logged_in"): return redirect(url_for("login")) return f(*args, **kwargs) return decorated app.secret_key = "connectgh_secret_2026_xK9mP2" # Login credentials — change these APP_USERNAME = "admin" APP_PASSWORD = "connectgh2026" def login_required(f): @wraps(f) def decorated(*args, **kwargs): if not session.get("logged_in"): return redirect(url_for("login")) return f(*args, **kwargs) return decorated # ───────────────────────────────────────────── # DATABASE # ───────────────────────────────────────────── def get_db(): conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row return conn def init_db(): with get_db() as db: db.executescript(""" CREATE TABLE IF NOT EXISTS incidents ( id INTEGER PRIMARY KEY AUTOINCREMENT, ref TEXT UNIQUE NOT NULL, title TEXT NOT NULL, severity TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'Open', category TEXT NOT NULL, reported_by TEXT NOT NULL, assigned_to TEXT, location TEXT, affected TEXT, description TEXT, root_cause TEXT, resolution TEXT, start_time TEXT, end_time TEXT, created_at TEXT DEFAULT (datetime('now')), updated_at TEXT DEFAULT (datetime('now')) ); CREATE TABLE IF NOT EXISTS timeline ( id INTEGER PRIMARY KEY AUTOINCREMENT, incident_id INTEGER NOT NULL, timestamp TEXT DEFAULT (datetime('now')), action TEXT NOT NULL, actor TEXT NOT NULL, note TEXT, FOREIGN KEY(incident_id) REFERENCES incidents(id) ); """) def next_ref(): year = datetime.now().year with get_db() as db: row = db.execute("SELECT COUNT(*) as c FROM incidents WHERE ref LIKE ?", (f"INC-{year}-%",)).fetchone() return f"INC-{year}-{row['c']+1:04d}" def get_stats(): with get_db() as db: return { "total": db.execute("SELECT COUNT(*) FROM incidents").fetchone()[0], "open": db.execute("SELECT COUNT(*) FROM incidents WHERE status='Open'").fetchone()[0], "progress": db.execute("SELECT COUNT(*) FROM incidents WHERE status='In Progress'").fetchone()[0], "resolved": db.execute("SELECT COUNT(*) FROM incidents WHERE status='Resolved'").fetchone()[0], "closed": db.execute("SELECT COUNT(*) FROM incidents WHERE status='Closed'").fetchone()[0], "critical": db.execute("SELECT COUNT(*) FROM incidents WHERE severity='Critical'").fetchone()[0], "high": db.execute("SELECT COUNT(*) FROM incidents WHERE severity='High'").fetchone()[0], "medium": db.execute("SELECT COUNT(*) FROM incidents WHERE severity='Medium'").fetchone()[0], "low": db.execute("SELECT COUNT(*) FROM incidents WHERE severity='Low'").fetchone()[0], } # ───────────────────────────────────────────── # ROUTES # ───────────────────────────────────────────── @app.route("/") @login_required def dashboard(): with get_db() as db: incidents = db.execute("SELECT * FROM incidents ORDER BY created_at DESC LIMIT 10").fetchall() recent_tl = db.execute(""" SELECT t.*, i.ref, i.title FROM timeline t JOIN incidents i ON i.id = t.incident_id ORDER BY t.timestamp DESC LIMIT 8 """).fetchall() return render_template("dashboard.html", incidents=incidents, stats=get_stats(), recent_tl=recent_tl) @app.route("/incidents") @login_required def incidents_list(): sev = request.args.get("sev", "") status = request.args.get("status", "") cat = request.args.get("cat", "") q = request.args.get("q", "") sql = "SELECT * FROM incidents WHERE 1=1" params = [] if sev: sql += " AND severity=?"; params.append(sev) if status: sql += " AND status=?"; params.append(status) if cat: sql += " AND category=?"; params.append(cat) if q: sql += " AND (title LIKE ? OR ref LIKE ? OR reported_by LIKE ?)"; params += [f"%{q}%"]*3 sql += " ORDER BY created_at DESC" with get_db() as db: incidents = db.execute(sql, params).fetchall() cats = [r[0] for r in db.execute("SELECT DISTINCT category FROM incidents ORDER BY category").fetchall()] return render_template("incidents.html", incidents=incidents, stats=get_stats(), sev=sev, status=status, cat=cat, q=q, cats=cats) @app.route("/incident/new", methods=["GET","POST"]) @login_required def new_incident(): if request.method == "POST": f = request.form ref = next_ref() with get_db() as db: db.execute(""" INSERT INTO incidents (ref,title,severity,status,category,reported_by, assigned_to,location,affected,description,start_time) VALUES (?,?,?,?,?,?,?,?,?,?,?) """, (ref, f["title"], f["severity"], "Open", f["category"], f["reported_by"], f.get("assigned_to"), f.get("location"), f.get("affected"), f.get("description"), f.get("start_time"))) iid = db.execute("SELECT id FROM incidents WHERE ref=?",(ref,)).fetchone()["id"] db.execute("INSERT INTO timeline (incident_id,action,actor,note) VALUES (?,?,?,?)", (iid, "Incident created", f["reported_by"], f.get("description",""))) return redirect(url_for("view_incident", ref=ref)) return render_template("form.html", incident=None, action="new") @app.route("/incident/") @login_required def view_incident(ref): with get_db() as db: inc = db.execute("SELECT * FROM incidents WHERE ref=?", (ref,)).fetchone() if not inc: return render_template("404.html"), 404 tl = db.execute("SELECT * FROM timeline WHERE incident_id=? ORDER BY timestamp", (inc["id"],)).fetchall() return render_template("detail.html", inc=inc, timeline=tl) @app.route("/incident//edit", methods=["GET","POST"]) @login_required def edit_incident(ref): with get_db() as db: inc = db.execute("SELECT * FROM incidents WHERE ref=?", (ref,)).fetchone() if not inc: return render_template("404.html"), 404 if request.method == "POST": f = request.form db.execute("""UPDATE incidents SET title=?,severity=?,status=?,category=?, assigned_to=?,location=?,affected=?,description=?,root_cause=?, resolution=?,start_time=?,end_time=?,updated_at=datetime('now') WHERE ref=?""", (f["title"],f["severity"],f["status"],f["category"], f.get("assigned_to"),f.get("location"),f.get("affected"), f.get("description"),f.get("root_cause"),f.get("resolution"), f.get("start_time"),f.get("end_time"),ref)) db.execute("INSERT INTO timeline (incident_id,action,actor,note) VALUES (?,?,?,?)", (inc["id"], f"Status → {f['status']}", f.get("assigned_to","System"), f.get("resolution") or f.get("description",""))) return redirect(url_for("view_incident", ref=ref)) return render_template("form.html", incident=inc, action="edit") @app.route("/incident//update-status", methods=["POST"]) @login_required def update_status(ref): data = request.get_json() with get_db() as db: inc = db.execute("SELECT id FROM incidents WHERE ref=?", (ref,)).fetchone() db.execute("UPDATE incidents SET status=?,updated_at=datetime('now') WHERE ref=?",(data["status"],ref)) db.execute("INSERT INTO timeline (incident_id,action,actor,note) VALUES (?,?,?,?)", (inc["id"], f"Status → {data['status']}", data.get("actor","System"), data.get("note",""))) return jsonify(ok=True) @app.route("/incident//add-note", methods=["POST"]) @login_required def add_note(ref): f = request.form with get_db() as db: inc = db.execute("SELECT id FROM incidents WHERE ref=?", (ref,)).fetchone() db.execute("INSERT INTO timeline (incident_id,action,actor,note) VALUES (?,?,?,?)", (inc["id"], "Note added", f["actor"], f["note"])) return redirect(url_for("view_incident", ref=ref)) @app.route("/incident//delete", methods=["POST"]) @login_required def delete_incident(ref): with get_db() as db: inc = db.execute("SELECT id FROM incidents WHERE ref=?", (ref,)).fetchone() if inc: db.execute("DELETE FROM timeline WHERE incident_id=?", (inc["id"],)) db.execute("DELETE FROM incidents WHERE ref=?", (ref,)) return redirect(url_for("incidents_list")) @app.route("/incident//report") @login_required def generate_report(ref): with get_db() as db: inc = db.execute("SELECT * FROM incidents WHERE ref=?", (ref,)).fetchone() tl = db.execute("SELECT * FROM timeline WHERE incident_id=? ORDER BY timestamp", (inc["id"],)).fetchall() path = os.path.join(RPT_DIR, f"{ref}.pdf") _build_pdf(dict(inc), [dict(t) for t in tl], path) return send_file(path, as_attachment=True, download_name=f"{ref}_ConnectGH_Report.pdf") @app.route("/analytics") @login_required def analytics(): with get_db() as db: by_sev = db.execute("SELECT severity, COUNT(*) as c FROM incidents GROUP BY severity").fetchall() by_cat = db.execute("SELECT category, COUNT(*) as c FROM incidents GROUP BY category ORDER BY c DESC").fetchall() by_stat = db.execute("SELECT status, COUNT(*) as c FROM incidents GROUP BY status").fetchall() by_day = db.execute(""" SELECT substr(created_at,1,10) as day, COUNT(*) as c FROM incidents WHERE created_at >= date('now','-30 days') GROUP BY day ORDER BY day """).fetchall() mttr_rows = db.execute(""" SELECT ref, round((julianday(end_time) - julianday(start_time))*24, 1) as hours FROM incidents WHERE end_time IS NOT NULL AND start_time IS NOT NULL ORDER BY created_at DESC LIMIT 10 """).fetchall() return render_template("analytics.html", stats=get_stats(), by_sev=[dict(r) for r in by_sev], by_cat=[dict(r) for r in by_cat], by_stat=[dict(r) for r in by_stat], by_day=[dict(r) for r in by_day], mttr=[dict(r) for r in mttr_rows]) @app.route("/api/incidents") @login_required def api_incidents(): with get_db() as db: rows = db.execute("SELECT * FROM incidents ORDER BY created_at DESC").fetchall() return jsonify([dict(r) for r in rows]) @app.route("/api/stats") @login_required def api_stats(): return jsonify(get_stats()) # ───────────────────────────────────────────── # PDF BUILDER # ───────────────────────────────────────────── BRAND = colors.HexColor("#1B4332") GOLD = colors.HexColor("#C8971A") DARK = colors.HexColor("#111827") MUTED = colors.HexColor("#6B7280") LOGO_PATH = os.path.join(BASE_DIR, "static", "connectgh_logo.png") SEV_COLORS = { "Critical": colors.HexColor("#DC2626"), "High": colors.HexColor("#EA580C"), "Medium": colors.HexColor("#D97706"), "Low": colors.HexColor("#16A34A"), } def _build_pdf(inc, timeline, path): doc = SimpleDocTemplate(path, pagesize=A4, leftMargin=2*cm, rightMargin=2*cm, topMargin=1.8*cm, bottomMargin=2*cm) body = [] # Logo + title header logo_cell = RLImage(LOGO_PATH, width=5.8*cm, height=1.64*cm) if os.path.exists(LOGO_PATH) \ else Paragraph("ConnectGH Limited", ParagraphStyle("l", fontName="Helvetica-Bold", fontSize=14)) hdr_data = [[logo_cell, Paragraph( f'NETWORK INCIDENT REPORT
' f'{inc["ref"]} · Generated {datetime.now().strftime("%d %b %Y %H:%M")}', ParagraphStyle("hd", fontName="Helvetica-Bold", fontSize=15, leading=22, alignment=2)) ]] hdr_t = Table(hdr_data, colWidths=[7*cm, 10*cm]) hdr_t.setStyle(TableStyle([ ("VALIGN", (0,0),(-1,-1), "MIDDLE"), ("BACKGROUND", (0,0),(-1,-1), colors.HexColor("#F0F7F4")), ("ROWPADDING", (0,0),(-1,-1), 10), ("LINEBELOW", (0,0),(-1,-1), 2, BRAND), ])) body.append(hdr_t) body.append(Spacer(1, 0.4*cm)) # Severity strip sev_col = SEV_COLORS.get(inc["severity"], BRAND) body.append(Table([[ Paragraph(f'{inc["severity"].upper()} SEVERITY', ParagraphStyle("sv", fontName="Helvetica-Bold", fontSize=10, textColor=colors.white)), Paragraph(f'Status: {inc["status"]}', ParagraphStyle("st", fontName="Helvetica", fontSize=10, textColor=colors.white)), Paragraph(f'Category: {inc["category"]}', ParagraphStyle("ct", fontName="Helvetica", fontSize=10, textColor=colors.white)), ]], colWidths=[5*cm, 5*cm, 7*cm], style=[ ("BACKGROUND",(0,0),(-1,-1), sev_col), ("ROWPADDING",(0,0),(-1,-1), 8), ("VALIGN", (0,0),(-1,-1), "MIDDLE"), ])) body.append(Spacer(1, 0.4*cm)) # Meta grid meta = [ ["Ref", inc["ref"], "Reported by", inc["reported_by"]], ["Assigned to", inc.get("assigned_to") or "—","Location", inc.get("location") or "—"], ["Affected", inc.get("affected") or "—", "Start time", inc.get("start_time") or "—"], ["End time", inc.get("end_time") or "—", "Last updated", inc.get("updated_at","")[:16]], ] t = Table(meta, colWidths=[3.5*cm,6*cm,3.5*cm,6*cm]) t.setStyle(TableStyle([ ("FONTNAME", (0,0),(-1,-1), "Helvetica"), ("FONTSIZE", (0,0),(-1,-1), 10), ("FONTNAME", (0,0),(0,-1), "Helvetica-Bold"), ("FONTNAME", (2,0),(2,-1), "Helvetica-Bold"), ("TEXTCOLOR", (0,0),(0,-1), BRAND), ("TEXTCOLOR", (2,0),(2,-1), BRAND), ("BACKGROUND", (0,0),(-1,-1), colors.HexColor("#F9FAFB")), ("BACKGROUND", (0,0),(0,-1), colors.HexColor("#E8F4EE")), ("BACKGROUND", (2,0),(2,-1), colors.HexColor("#E8F4EE")), ("GRID", (0,0),(-1,-1), 0.5, colors.HexColor("#D1E7DC")), ("ROWPADDING", (0,0),(-1,-1), 6), ])) body.append(t) body.append(Spacer(1, 0.5*cm)) def section(title, content): body.append(Paragraph(title, ParagraphStyle("sh", fontName="Helvetica-Bold", fontSize=11, textColor=BRAND, spaceBefore=10, spaceAfter=4))) body.append(HRFlowable(width="100%", thickness=1, color=GOLD)) body.append(Paragraph(content or "—", ParagraphStyle("bd", fontName="Helvetica", fontSize=10, leading=15, spaceBefore=5, spaceAfter=8, textColor=colors.HexColor("#374151")))) section("Description", inc.get("description")) section("Root Cause Analysis", inc.get("root_cause")) section("Resolution / Actions Taken", inc.get("resolution")) # Timeline table body.append(Paragraph("Incident Timeline", ParagraphStyle("sh2", fontName="Helvetica-Bold", fontSize=11, textColor=BRAND, spaceBefore=10, spaceAfter=4))) body.append(HRFlowable(width="100%", thickness=1, color=GOLD)) body.append(Spacer(1, 0.2*cm)) if timeline: tl_data = [["Timestamp","Action","Actor","Note"]] for e in timeline: tl_data.append([e["timestamp"][:16], e["action"], e["actor"], e.get("note") or ""]) tl_t = Table(tl_data, colWidths=[3.5*cm,3.5*cm,3*cm,9*cm]) tl_t.setStyle(TableStyle([ ("FONTNAME", (0,0),(-1,0), "Helvetica-Bold"), ("FONTNAME", (0,1),(-1,-1), "Helvetica"), ("FONTSIZE", (0,0),(-1,-1), 9), ("BACKGROUND", (0,0),(-1,0), BRAND), ("TEXTCOLOR", (0,0),(-1,0), colors.white), ("ROWBACKGROUNDS", (0,1),(-1,-1), [colors.white, colors.HexColor("#F0F7F4")]), ("GRID", (0,0),(-1,-1), 0.3, colors.HexColor("#D1E7DC")), ("ROWPADDING", (0,0),(-1,-1), 5), ("VALIGN", (0,0),(-1,-1), "TOP"), ])) body.append(tl_t) body.append(Spacer(1, 1*cm)) body.append(HRFlowable(width="100%", thickness=0.5, color=colors.HexColor("#C6DDD5"))) body.append(Paragraph("Confidential — ConnectGH Limited | Network Incident Management System", ParagraphStyle("ft", fontName="Helvetica", fontSize=8, textColor=MUTED, spaceBefore=4))) doc.build(body) @app.route("/login", methods=["GET", "POST"]) def login(): error = None if request.method == "POST": if request.form["username"] == APP_USERNAME and request.form["password"] == APP_PASSWORD: session["logged_in"] = True return redirect(url_for("dashboard")) error = "Invalid username or password" return render_template("login.html", error=error) @app.route("/logout") def logout(): session.clear() return redirect(url_for("login")) if __name__ == "__main__": init_db() url = "http://127.0.0.1:5050" threading.Timer(1.2, lambda: webbrowser.open(url)).start() print(f"\n ConnectGH Network Incident System → {url}\n Press Ctrl+C to stop.\n") app.run(port=5050, debug=False)