feat(reports): add Envelope, Phone Book (address+phone) and Rolodex Info reports
- PDF builders in app/reporting.py (envelope, phone+address, rolodex info) - Endpoints in app/main.py with auth, filtering, logging, Content-Disposition - New HTML template report_phone_book_address.html - Rolodex bulk actions updated with buttons/links - JS helper to submit selections to alternate endpoints Tested via docker compose build/up and health check.
This commit is contained in:
HotSwapp
224
app/main.py
224
app/main.py
@@ -31,7 +31,13 @@ from structlog import contextvars as structlog_contextvars
|
||||
from .database import create_tables, get_db, get_database_url
|
||||
from .models import User, Case, Client, Phone, Transaction, Document, Payment, ImportLog
|
||||
from .auth import authenticate_user, get_current_user_from_session
|
||||
from .reporting import build_phone_book_pdf, build_payments_detailed_pdf
|
||||
from .reporting import (
|
||||
build_phone_book_pdf,
|
||||
build_payments_detailed_pdf,
|
||||
build_envelope_pdf,
|
||||
build_phone_book_address_pdf,
|
||||
build_rolodex_info_pdf,
|
||||
)
|
||||
from .logging_config import setup_logging
|
||||
from .schemas import (
|
||||
ClientOut,
|
||||
@@ -2420,6 +2426,222 @@ async def payments_detailed_report(
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
# ------------------------------
|
||||
# Reports: Phone Book (Address + Phone)
|
||||
# ------------------------------
|
||||
|
||||
@app.post("/reports/phone-book-address")
|
||||
async def phone_book_address_post(request: Request):
|
||||
"""Accept selected client IDs from forms and redirect to GET for rendering."""
|
||||
user = get_current_user_from_session(request.session)
|
||||
if not user:
|
||||
return RedirectResponse(url="/login", status_code=302)
|
||||
|
||||
form = await request.form()
|
||||
client_ids = form.getlist("client_ids")
|
||||
if not client_ids:
|
||||
return RedirectResponse(url="/rolodex", status_code=302)
|
||||
|
||||
ids_param = "&".join([f"client_ids={cid}" for cid in client_ids])
|
||||
return RedirectResponse(url=f"/reports/phone-book-address?{ids_param}", status_code=302)
|
||||
|
||||
|
||||
@app.get("/reports/phone-book-address")
|
||||
async def phone_book_address_report(
|
||||
request: Request,
|
||||
client_ids: List[int] | None = Query(None),
|
||||
q: str | None = Query(None, description="Filter by name/company"),
|
||||
phone: str | None = Query(None, description="Phone contains"),
|
||||
format: str | None = Query(None, description="csv or pdf for export"),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
user = get_current_user_from_session(request.session)
|
||||
if not user:
|
||||
return RedirectResponse(url="/login", status_code=302)
|
||||
|
||||
query = db.query(Client).options(joinedload(Client.phones))
|
||||
if client_ids:
|
||||
query = query.filter(Client.id.in_(client_ids))
|
||||
else:
|
||||
if q:
|
||||
like = f"%{q}%"
|
||||
query = query.filter(
|
||||
or_(Client.first_name.ilike(like), Client.last_name.ilike(like), Client.company.ilike(like))
|
||||
)
|
||||
if phone:
|
||||
query = query.filter(Client.phones.any(Phone.phone_number.ilike(f"%{phone}%")))
|
||||
|
||||
clients = query.order_by(Client.last_name.asc().nulls_last(), Client.first_name.asc().nulls_last()).all()
|
||||
|
||||
if format == "csv":
|
||||
# Build CSV output
|
||||
output = StringIO()
|
||||
writer = csv.writer(output)
|
||||
writer.writerow(["Last", "First", "Company", "Address", "City", "State", "ZIP", "Phone Type", "Phone Number"])
|
||||
for c in clients:
|
||||
if c.phones:
|
||||
for p in c.phones:
|
||||
writer.writerow([
|
||||
c.last_name or "",
|
||||
c.first_name or "",
|
||||
c.company or "",
|
||||
c.address or "",
|
||||
c.city or "",
|
||||
c.state or "",
|
||||
c.zip_code or "",
|
||||
p.phone_type or "",
|
||||
p.phone_number or "",
|
||||
])
|
||||
else:
|
||||
writer.writerow([
|
||||
c.last_name or "",
|
||||
c.first_name or "",
|
||||
c.company or "",
|
||||
c.address or "",
|
||||
c.city or "",
|
||||
c.state or "",
|
||||
c.zip_code or "",
|
||||
"",
|
||||
"",
|
||||
])
|
||||
csv_bytes = output.getvalue().encode("utf-8")
|
||||
return Response(
|
||||
content=csv_bytes,
|
||||
media_type="text/csv",
|
||||
headers={"Content-Disposition": "attachment; filename=phone_book_address.csv"},
|
||||
)
|
||||
|
||||
if format == "pdf":
|
||||
pdf_bytes = build_phone_book_address_pdf(clients)
|
||||
return Response(
|
||||
content=pdf_bytes,
|
||||
media_type="application/pdf",
|
||||
headers={"Content-Disposition": "attachment; filename=phone_book_address.pdf"},
|
||||
)
|
||||
|
||||
logger.info("phone_book_address_render", count=len(clients))
|
||||
return templates.TemplateResponse(
|
||||
"report_phone_book_address.html",
|
||||
{
|
||||
"request": request,
|
||||
"user": user,
|
||||
"clients": clients,
|
||||
"q": q,
|
||||
"phone": phone,
|
||||
"client_ids": client_ids or [],
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
# ------------------------------
|
||||
# Reports: Envelope (PDF)
|
||||
# ------------------------------
|
||||
|
||||
@app.post("/reports/envelope")
|
||||
async def envelope_report_post(request: Request):
|
||||
"""Accept selected client IDs and redirect to GET for PDF download."""
|
||||
user = get_current_user_from_session(request.session)
|
||||
if not user:
|
||||
return RedirectResponse(url="/login", status_code=302)
|
||||
|
||||
form = await request.form()
|
||||
client_ids = form.getlist("client_ids")
|
||||
if not client_ids:
|
||||
return RedirectResponse(url="/rolodex", status_code=302)
|
||||
|
||||
ids_param = "&".join([f"client_ids={cid}" for cid in client_ids])
|
||||
return RedirectResponse(url=f"/reports/envelope?{ids_param}&format=pdf", status_code=302)
|
||||
|
||||
|
||||
@app.get("/reports/envelope")
|
||||
async def envelope_report(
|
||||
request: Request,
|
||||
client_ids: List[int] | None = Query(None),
|
||||
q: str | None = Query(None, description="Filter by name/company"),
|
||||
phone: str | None = Query(None, description="Phone contains (optional)"),
|
||||
format: str | None = Query("pdf", description="pdf output only"),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
user = get_current_user_from_session(request.session)
|
||||
if not user:
|
||||
return RedirectResponse(url="/login", status_code=302)
|
||||
|
||||
query = db.query(Client)
|
||||
if client_ids:
|
||||
query = query.filter(Client.id.in_(client_ids))
|
||||
else:
|
||||
if q:
|
||||
like = f"%{q}%"
|
||||
query = query.filter(
|
||||
or_(Client.first_name.ilike(like), Client.last_name.ilike(like), Client.company.ilike(like))
|
||||
)
|
||||
if phone:
|
||||
# include clients that have a matching phone
|
||||
query = query.join(Phone, isouter=True).filter(or_(Phone.phone_number.ilike(f"%{phone}%"), Phone.id == None)).distinct() # noqa: E711
|
||||
|
||||
clients = query.order_by(Client.last_name.asc().nulls_last(), Client.first_name.asc().nulls_last()).all()
|
||||
|
||||
# Always produce PDF
|
||||
pdf_bytes = build_envelope_pdf(clients)
|
||||
logger.info("envelope_pdf", count=len(clients))
|
||||
return Response(
|
||||
content=pdf_bytes,
|
||||
media_type="application/pdf",
|
||||
headers={"Content-Disposition": "attachment; filename=envelopes.pdf"},
|
||||
)
|
||||
|
||||
|
||||
# ------------------------------
|
||||
# Reports: Rolodex Info (PDF)
|
||||
# ------------------------------
|
||||
|
||||
@app.post("/reports/rolodex-info")
|
||||
async def rolodex_info_post(request: Request):
|
||||
user = get_current_user_from_session(request.session)
|
||||
if not user:
|
||||
return RedirectResponse(url="/login", status_code=302)
|
||||
|
||||
form = await request.form()
|
||||
client_ids = form.getlist("client_ids")
|
||||
if not client_ids:
|
||||
return RedirectResponse(url="/rolodex", status_code=302)
|
||||
|
||||
ids_param = "&".join([f"client_ids={cid}" for cid in client_ids])
|
||||
return RedirectResponse(url=f"/reports/rolodex-info?{ids_param}&format=pdf", status_code=302)
|
||||
|
||||
|
||||
@app.get("/reports/rolodex-info")
|
||||
async def rolodex_info_report(
|
||||
request: Request,
|
||||
client_ids: List[int] | None = Query(None),
|
||||
q: str | None = Query(None, description="Filter by name/company"),
|
||||
format: str | None = Query("pdf", description="pdf output only"),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
user = get_current_user_from_session(request.session)
|
||||
if not user:
|
||||
return RedirectResponse(url="/login", status_code=302)
|
||||
|
||||
query = db.query(Client).options(joinedload(Client.phones))
|
||||
if client_ids:
|
||||
query = query.filter(Client.id.in_(client_ids))
|
||||
elif q:
|
||||
like = f"%{q}%"
|
||||
query = query.filter(
|
||||
or_(Client.first_name.ilike(like), Client.last_name.ilike(like), Client.company.ilike(like))
|
||||
)
|
||||
|
||||
clients = query.order_by(Client.last_name.asc().nulls_last(), Client.first_name.asc().nulls_last()).all()
|
||||
|
||||
pdf_bytes = build_rolodex_info_pdf(clients)
|
||||
logger.info("rolodex_info_pdf", count=len(clients))
|
||||
return Response(
|
||||
content=pdf_bytes,
|
||||
media_type="application/pdf",
|
||||
headers={"Content-Disposition": "attachment; filename=rolodex_info.pdf"},
|
||||
)
|
||||
|
||||
# ------------------------------
|
||||
# JSON API: list/filter endpoints
|
||||
# ------------------------------
|
||||
|
||||
Reference in New Issue
Block a user