nikola
147 lines
4.5 KiB
Python
147 lines
4.5 KiB
Python
import csv
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from flask import Flask, redirect, render_template, request, send_file, url_for
|
|
|
|
from app.diagram import render_diagram
|
|
from app.explorer import render_explorer
|
|
from app.parser import parse_gam_raw_csv, sharing_matrix, summarize, top_stats
|
|
|
|
BASE_DIR = Path(__file__).resolve().parent.parent
|
|
DATA_DIR = BASE_DIR / "data"
|
|
DATA_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
app = Flask(
|
|
__name__,
|
|
template_folder=str(BASE_DIR / "templates"),
|
|
static_folder=str(BASE_DIR / "static"),
|
|
)
|
|
app.config["MAX_CONTENT_LENGTH"] = 50 * 1024 * 1024
|
|
|
|
STATE = {
|
|
"last_job": None,
|
|
"last_error": None,
|
|
}
|
|
|
|
|
|
def write_csv(path: Path, rows: list[dict]) -> None:
|
|
if not rows:
|
|
return
|
|
with path.open("w", newline="", encoding="utf-8") as f:
|
|
w = csv.DictWriter(f, fieldnames=list(rows[0].keys()))
|
|
w.writeheader()
|
|
w.writerows(rows)
|
|
|
|
|
|
@app.get("/")
|
|
def index():
|
|
last_job = STATE["last_job"]
|
|
context = {
|
|
"has_data": False,
|
|
"stats": {},
|
|
"diagram_path": None,
|
|
"cleaned_path": None,
|
|
"summary_path": None,
|
|
"sharing_matrix_path": None,
|
|
"explorer_path": None,
|
|
"explorer_url": None,
|
|
"build_id": None,
|
|
"last_upload_time": None,
|
|
"process_ms": None,
|
|
"curated_count": None,
|
|
"error": STATE.get("last_error"),
|
|
}
|
|
|
|
if last_job:
|
|
context.update(last_job)
|
|
context["has_data"] = True
|
|
|
|
return render_template("index.html", **context)
|
|
|
|
|
|
@app.post("/upload")
|
|
def upload_csv():
|
|
upload = request.files.get("csv_file")
|
|
if not upload or not upload.filename:
|
|
return redirect(url_for("index"))
|
|
|
|
try:
|
|
t0 = datetime.now(timezone.utc)
|
|
job_id = uuid.uuid4().hex[:10]
|
|
job_dir = DATA_DIR / job_id
|
|
job_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
raw_path = job_dir / "gdrive_test_raw.csv"
|
|
upload.save(raw_path)
|
|
|
|
curated_rows = parse_gam_raw_csv(raw_path)
|
|
summary_rows = summarize(curated_rows)
|
|
sharing_rows = sharing_matrix(curated_rows)
|
|
stats = top_stats(curated_rows)
|
|
|
|
cleaned_csv = job_dir / "gdrive_curated.csv"
|
|
summary_csv = job_dir / "gdrive_summary.csv"
|
|
sharing_csv = job_dir / "gdrive_sharing_matrix.csv"
|
|
diagram_html = job_dir / "diagram.html"
|
|
explorer_html = job_dir / "explorer.html"
|
|
|
|
write_csv(cleaned_csv, curated_rows)
|
|
write_csv(summary_csv, summary_rows)
|
|
write_csv(sharing_csv, sharing_rows)
|
|
render_diagram(summary_rows, diagram_html)
|
|
build_id = datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S")
|
|
render_explorer(curated_rows, explorer_html, build_id=build_id)
|
|
t1 = datetime.now(timezone.utc)
|
|
process_ms = int((t1 - t0).total_seconds() * 1000)
|
|
|
|
STATE["last_job"] = {
|
|
"stats": stats,
|
|
"diagram_path": f"/data/{job_id}/diagram.html",
|
|
"cleaned_path": f"/download/{job_id}/cleaned",
|
|
"summary_path": f"/download/{job_id}/summary",
|
|
"sharing_matrix_path": f"/download/{job_id}/sharing-matrix",
|
|
"explorer_path": f"/data/{job_id}/explorer.html",
|
|
"explorer_url": f"/data/{job_id}/explorer.html?v={build_id}",
|
|
"build_id": build_id,
|
|
"last_upload_time": t1.strftime("%Y-%m-%d %H:%M:%S UTC"),
|
|
"process_ms": process_ms,
|
|
"curated_count": len(curated_rows),
|
|
"job_id": job_id,
|
|
}
|
|
STATE["last_error"] = None
|
|
except Exception as exc:
|
|
STATE["last_error"] = (
|
|
"Upload failed. Check CSV format and try again. "
|
|
f"Error: {exc}"
|
|
)
|
|
|
|
return redirect(url_for("index"))
|
|
|
|
|
|
@app.get("/download/<job_id>/<kind>")
|
|
def download(job_id: str, kind: str):
|
|
job_dir = DATA_DIR / job_id
|
|
if kind == "cleaned":
|
|
return send_file(job_dir / "gdrive_curated.csv", as_attachment=True)
|
|
if kind == "summary":
|
|
return send_file(job_dir / "gdrive_summary.csv", as_attachment=True)
|
|
if kind == "sharing-matrix":
|
|
return send_file(job_dir / "gdrive_sharing_matrix.csv", as_attachment=True)
|
|
return redirect(url_for("index"))
|
|
|
|
|
|
@app.get("/data/<job_id>/diagram.html")
|
|
def serve_diagram(job_id: str):
|
|
return send_file(DATA_DIR / job_id / "diagram.html")
|
|
|
|
|
|
@app.get("/data/<job_id>/explorer.html")
|
|
def serve_explorer(job_id: str):
|
|
return send_file(DATA_DIR / job_id / "explorer.html")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app.run(host="0.0.0.0", port=8080, debug=False)
|