367 lines
11 KiB
Python
367 lines
11 KiB
Python
import csv
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
|
|
from fastapi import APIRouter, File, HTTPException, UploadFile
|
|
from fastapi.responses import FileResponse
|
|
from pydantic import BaseModel
|
|
|
|
from tools.doc_generator.logic.document_types import get_document_type, list_document_types
|
|
from tools.doc_generator.logic.renderer import generate_docx
|
|
from tools.doc_generator.logic.excel_mapper import export_profile_excel, load_excel_maps, load_excel_map, resolve_excel_template
|
|
|
|
router = APIRouter()
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
|
|
EXPORTS_DIR = PROJECT_ROOT / "exports"
|
|
INPUTS_DIR = PROJECT_ROOT / "inputs"
|
|
UPLOADS_DIR = INPUTS_DIR / "uploads"
|
|
JSON_UPLOADS_DIR = UPLOADS_DIR / "json"
|
|
DATA_UPLOADS_DIR = UPLOADS_DIR / "data"
|
|
TEMPLATE_UPLOADS_DIR = UPLOADS_DIR / "templates"
|
|
|
|
|
|
class ExportExcelRequest(BaseModel):
|
|
document_type_id: str
|
|
data: dict
|
|
map_id: str = "legacy_datafile"
|
|
|
|
|
|
class GenerateDocRequest(BaseModel):
|
|
document_type_id: str
|
|
data: dict
|
|
template_id: str | None = None
|
|
|
|
|
|
def safe_filename(filename: str) -> str:
|
|
filename = Path(filename or "upload").name
|
|
filename = re.sub(r"[^A-Za-z0-9._ -]+", "", filename)
|
|
filename = re.sub(r"\s+", "_", filename).strip("._ ")
|
|
return filename or "upload"
|
|
|
|
|
|
def save_upload(upload: UploadFile, directory: Path) -> Path:
|
|
directory.mkdir(parents=True, exist_ok=True)
|
|
filename = safe_filename(upload.filename)
|
|
path = directory / filename
|
|
|
|
counter = 1
|
|
while path.exists():
|
|
stem = path.stem
|
|
suffix = path.suffix
|
|
path = directory / f"{stem}_{counter}{suffix}"
|
|
counter += 1
|
|
|
|
with path.open("wb") as f:
|
|
f.write(upload.file.read())
|
|
|
|
return path
|
|
|
|
|
|
@router.get("/document-types")
|
|
def document_types():
|
|
return {"document_types": list_document_types()}
|
|
|
|
|
|
@router.get("/document-types/{document_type_id}")
|
|
def document_type(document_type_id: str):
|
|
try:
|
|
return get_document_type(document_type_id)
|
|
except FileNotFoundError as exc:
|
|
raise HTTPException(status_code=404, detail=str(exc))
|
|
|
|
|
|
@router.post("/upload-json")
|
|
async def upload_json(file: UploadFile = File(...)):
|
|
if not file.filename.lower().endswith(".json"):
|
|
raise HTTPException(status_code=400, detail="Upload must be a JSON file.")
|
|
|
|
path = save_upload(file, JSON_UPLOADS_DIR)
|
|
|
|
try:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=400, detail=f"Invalid JSON: {exc}")
|
|
|
|
return {
|
|
"ok": True,
|
|
"filename": path.name,
|
|
"saved_path": str(path.relative_to(PROJECT_ROOT)),
|
|
"json": data,
|
|
}
|
|
|
|
|
|
@router.get("/uploaded-json")
|
|
def uploaded_json_files():
|
|
JSON_UPLOADS_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
files = []
|
|
for path in sorted(JSON_UPLOADS_DIR.glob("*.json")):
|
|
files.append({
|
|
"filename": path.name,
|
|
"saved_path": str(path.relative_to(PROJECT_ROOT)),
|
|
"modified": path.stat().st_mtime,
|
|
})
|
|
|
|
return {"files": files}
|
|
|
|
|
|
@router.get("/uploaded-json/{filename}")
|
|
def get_uploaded_json(filename: str):
|
|
file_path = JSON_UPLOADS_DIR / safe_filename(filename)
|
|
|
|
if not file_path.exists():
|
|
raise HTTPException(status_code=404, detail="Uploaded JSON not found.")
|
|
|
|
try:
|
|
data = json.loads(file_path.read_text(encoding="utf-8"))
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=400, detail=f"Invalid JSON: {exc}")
|
|
|
|
return {
|
|
"ok": True,
|
|
"filename": file_path.name,
|
|
"saved_path": str(file_path.relative_to(PROJECT_ROOT)),
|
|
"json": data,
|
|
}
|
|
|
|
|
|
@router.delete("/uploaded-json/{filename}")
|
|
def delete_uploaded_json(filename: str):
|
|
file_path = JSON_UPLOADS_DIR / safe_filename(filename)
|
|
|
|
if not file_path.exists():
|
|
raise HTTPException(status_code=404, detail="Uploaded JSON not found.")
|
|
|
|
file_path.unlink()
|
|
|
|
return {
|
|
"ok": True,
|
|
"deleted": file_path.name,
|
|
}
|
|
|
|
|
|
@router.post("/upload-data")
|
|
async def upload_data(file: UploadFile = File(...)):
|
|
if not file.filename.lower().endswith(".csv"):
|
|
raise HTTPException(status_code=400, detail="For now, upload data as CSV.")
|
|
|
|
path = save_upload(file, DATA_UPLOADS_DIR)
|
|
|
|
try:
|
|
with path.open("r", encoding="utf-8-sig", newline="") as f:
|
|
reader = csv.DictReader(f)
|
|
rows = list(reader)
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=400, detail=f"Could not read CSV: {exc}")
|
|
|
|
if not rows:
|
|
raise HTTPException(status_code=400, detail="CSV has no data rows.")
|
|
|
|
return {
|
|
"ok": True,
|
|
"filename": path.name,
|
|
"saved_path": str(path.relative_to(PROJECT_ROOT)),
|
|
"data": rows[0],
|
|
"row_count": len(rows),
|
|
}
|
|
|
|
|
|
|
|
|
|
@router.post("/upload-template/{document_type_id}")
|
|
async def upload_template(document_type_id: str, file: UploadFile = File(...)):
|
|
if not file.filename.lower().endswith(".docx"):
|
|
raise HTTPException(status_code=400, detail="Upload must be a DOCX template.")
|
|
|
|
# Validate profile exists.
|
|
try:
|
|
get_document_type(document_type_id)
|
|
except FileNotFoundError as exc:
|
|
raise HTTPException(status_code=404, detail=str(exc))
|
|
|
|
profile_dir = TEMPLATE_UPLOADS_DIR / safe_filename(document_type_id)
|
|
path = save_upload(file, profile_dir)
|
|
|
|
return {
|
|
"ok": True,
|
|
"filename": path.name,
|
|
"template_id": f"uploaded:{path.name}",
|
|
"saved_path": str(path.relative_to(PROJECT_ROOT)),
|
|
}
|
|
|
|
|
|
@router.get("/uploaded-templates/{document_type_id}")
|
|
def uploaded_templates(document_type_id: str):
|
|
profile_dir = TEMPLATE_UPLOADS_DIR / safe_filename(document_type_id)
|
|
profile_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
templates = []
|
|
for path in sorted(profile_dir.glob("*.docx")):
|
|
templates.append({
|
|
"id": f"uploaded:{path.name}",
|
|
"label": path.name,
|
|
"filename": path.name,
|
|
"saved_path": str(path.relative_to(PROJECT_ROOT)),
|
|
"modified": path.stat().st_mtime,
|
|
})
|
|
|
|
return {"templates": templates}
|
|
|
|
|
|
@router.delete("/uploaded-template/{document_type_id}/{template_id}")
|
|
def delete_uploaded_template(document_type_id: str, template_id: str):
|
|
doc_dir = TEMPLATE_UPLOADS_DIR / safe_filename(document_type_id)
|
|
file_path = doc_dir / safe_filename(template_id)
|
|
|
|
if not file_path.exists():
|
|
raise HTTPException(status_code=404, detail="Template not found.")
|
|
|
|
file_path.unlink()
|
|
return {"deleted": True, "template_id": file_path.name}
|
|
|
|
|
|
|
|
|
|
|
|
@router.get("/excel-maps/{document_type_id}")
|
|
def get_excel_maps_route(document_type_id: str):
|
|
try:
|
|
maps = load_excel_maps(document_type_id)
|
|
except FileNotFoundError:
|
|
return {"maps": []}
|
|
|
|
return {
|
|
"maps": [
|
|
{
|
|
"id": item.get("id"),
|
|
"label": item.get("label", item.get("id")),
|
|
"field_count": len(item.get("field_to_cell", {})),
|
|
"fields": sorted(item.get("field_to_cell", {}).keys()),
|
|
"template": item.get("template"),
|
|
"mode": item.get("mode", "map"),
|
|
}
|
|
for item in maps
|
|
]
|
|
}
|
|
|
|
@router.post("/export-excel")
|
|
def export_excel(request: ExportExcelRequest):
|
|
try:
|
|
output_path = export_profile_excel(
|
|
request.document_type_id,
|
|
request.data,
|
|
request.map_id,
|
|
)
|
|
return {
|
|
"filename": output_path.name,
|
|
"download_url": f"/api/doc-generator/download/{output_path.name}",
|
|
}
|
|
except FileNotFoundError as exc:
|
|
raise HTTPException(status_code=404, detail=str(exc))
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
|
|
|
|
@router.post("/generate")
|
|
def generate_document(request: GenerateDocRequest):
|
|
try:
|
|
output_path = generate_docx(request.document_type_id, request.data, request.template_id)
|
|
except FileNotFoundError as exc:
|
|
raise HTTPException(status_code=404, detail=str(exc))
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
|
|
return {
|
|
"ok": True,
|
|
"filename": output_path.name,
|
|
"download_url": f"/api/doc-generator/download/{output_path.name}"
|
|
}
|
|
|
|
|
|
|
|
|
|
@router.get("/excel-template/{template_name}")
|
|
def download_excel_template(template_name: str):
|
|
allowed = {
|
|
"datafile": "template.xlsx",
|
|
"amortization": "amortization.xlsx",
|
|
}
|
|
|
|
filename = allowed.get(template_name)
|
|
if not filename:
|
|
raise HTTPException(status_code=404, detail="Unknown Excel template.")
|
|
|
|
path = OLD_WORD_DOC_GENERATOR_DIR / filename
|
|
|
|
if not path.exists():
|
|
raise HTTPException(status_code=404, detail=f"Excel template not found: {path}")
|
|
|
|
return FileResponse(
|
|
path,
|
|
filename=filename,
|
|
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
@router.get("/excel-template-by-map/{document_type_id}/{map_id}")
|
|
def download_excel_template_by_map(document_type_id: str, map_id: str):
|
|
try:
|
|
excel_map = load_excel_map(document_type_id, map_id)
|
|
template_path = resolve_excel_template(excel_map.get("template"))
|
|
except FileNotFoundError as exc:
|
|
raise HTTPException(status_code=404, detail=str(exc))
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
|
|
return FileResponse(
|
|
path=template_path,
|
|
filename=template_path.name,
|
|
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
|
)
|
|
|
|
|
|
|
|
|
|
@router.delete("/uploaded-template/{document_type_id}/{filename}")
|
|
def delete_uploaded_template(document_type_id: str, filename: str):
|
|
folder = TEMPLATE_UPLOADS_DIR / safe_filename(document_type_id)
|
|
path = folder / safe_filename(filename)
|
|
|
|
if not path.exists():
|
|
raise HTTPException(status_code=404, detail="Uploaded template not found.")
|
|
|
|
if path.suffix.lower() != ".docx":
|
|
raise HTTPException(status_code=400, detail="Only DOCX templates can be deleted here.")
|
|
|
|
path.unlink()
|
|
return {"deleted": True, "filename": path.name}
|
|
|
|
@router.get("/download/{filename}")
|
|
def download_file(filename: str):
|
|
safe_name = safe_filename(filename)
|
|
path = EXPORTS_DIR / safe_name
|
|
|
|
if not path.exists():
|
|
raise HTTPException(status_code=404, detail="File not found.")
|
|
|
|
suffix = path.suffix.lower()
|
|
|
|
media_types = {
|
|
".docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
|
|
".xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
|
".csv": "text/csv",
|
|
".json": "application/json",
|
|
".pdf": "application/pdf",
|
|
}
|
|
|
|
return FileResponse(
|
|
path,
|
|
filename=path.name,
|
|
media_type=media_types.get(suffix, "application/octet-stream"),
|
|
)
|