A R R R Associates
494 lines
21 KiB
Python
494 lines
21 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import date, datetime, timezone
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
from fastapi import UploadFile
|
|
from sqlalchemy import and_, func, or_, select
|
|
from sqlalchemy.orm import Session, joinedload
|
|
|
|
from app.core.settings import get_settings
|
|
from app.modules.clients.models import Client
|
|
from app.modules.core.tenancy.settings_models import BranchSettings
|
|
from app.modules.core.rbac.deps import get_user_permissions, get_user_roles
|
|
from app.modules.notice_cases.models import (
|
|
NoticeCase,
|
|
NoticeCaseDocument,
|
|
NoticeCaseEvent,
|
|
NoticeCaseHearing,
|
|
NoticeCaseOrder,
|
|
)
|
|
from app.modules.services.models import ClientServiceSubscription
|
|
|
|
DEPARTMENTS = ["GST", "Income Tax", "ROC", "PF", "ESI", "Labour", "MSME", "Other"]
|
|
CASE_TYPES = ["Notice", "Assessment", "Appeal", "Rectification", "Refund", "Registration", "Investigation", "Other"]
|
|
CASE_STATUSES = ["open", "reply_pending", "reply_filed", "hearing_scheduled", "order_received", "appeal_pending", "appeal_filed", "closed", "archived"]
|
|
CASE_PRIORITIES = ["low", "normal", "high", "urgent"]
|
|
EVENT_TYPES = ["Notice Received", "Reply Filed", "Hearing", "Order Received", "Appeal Filed", "Rectification Filed", "Payment Made", "Internal Note", "Client Clarification", "Other"]
|
|
HEARING_STATUSES = ["scheduled", "attended", "adjourned", "missed", "cancelled"]
|
|
ORDER_TYPES = ["Assessment Order", "Appeal Order", "Rectification Order", "Refund Order", "Penalty Order", "Other"]
|
|
CASE_DOCUMENT_TYPES = ["NOTICE", "REPLY", "APPEAL", "ORDER", "CHALLAN", "WORKING", "CLIENT_DOCUMENT", "ACKNOWLEDGEMENT", "OTHER"]
|
|
|
|
|
|
def _has(permissions: list[str], code: str) -> bool:
|
|
return code in set(permissions or [])
|
|
|
|
|
|
def can_view_notice_cases(db: Session, user) -> bool:
|
|
perms = get_user_permissions(db, user.id)
|
|
return _has(perms, "notice_cases.view")
|
|
|
|
|
|
def can_manage_notice_cases(db: Session, user) -> bool:
|
|
perms = get_user_permissions(db, user.id)
|
|
return _has(perms, "notice_cases.create") or _has(perms, "notice_cases.edit")
|
|
|
|
|
|
def can_upload_notice_case_documents(db: Session, user) -> bool:
|
|
perms = get_user_permissions(db, user.id)
|
|
return _has(perms, "notice_cases.documents.upload")
|
|
|
|
|
|
def can_delete_notice_case_documents(db: Session, user) -> bool:
|
|
perms = get_user_permissions(db, user.id)
|
|
return _has(perms, "notice_cases.documents.delete")
|
|
|
|
|
|
def user_can_access_case(
|
|
db: Session,
|
|
user,
|
|
case: NoticeCase,
|
|
*,
|
|
active_tenant_id: int | None,
|
|
active_branch_id: int | None,
|
|
active_financial_year: str | None = None,
|
|
active_assessment_year: str | None = None,
|
|
) -> bool:
|
|
roles = set(get_user_roles(db, user.id))
|
|
perms = set(get_user_permissions(db, user.id))
|
|
if "System Admin" in roles and "notice_cases.cross_tenant" in perms:
|
|
return active_tenant_id in (None, case.tenant_id) or True
|
|
if case.tenant_id != int(active_tenant_id or getattr(user, "tenant_id", 0) or 0):
|
|
return False
|
|
if active_branch_id and case.branch_id and case.branch_id != int(active_branch_id):
|
|
return False
|
|
if "notice_cases.view" not in perms:
|
|
return False
|
|
fy = (active_financial_year or "").strip()
|
|
ay = (active_assessment_year or "").strip()
|
|
if fy and (case.financial_year or "").strip() and (case.financial_year or "").strip() != fy:
|
|
return False
|
|
if ay and not fy and (case.assessment_year or "").strip() and (case.assessment_year or "").strip() != ay:
|
|
return False
|
|
if "notice_cases.view.own_only" in perms and not ({case.assigned_partner_user_id, case.assigned_manager_user_id, case.assigned_staff_user_id} & {user.id}):
|
|
return False
|
|
return True
|
|
|
|
|
|
def parse_date(value: str | None) -> date | None:
|
|
value = (value or "").strip()
|
|
if not value:
|
|
return None
|
|
return date.fromisoformat(value)
|
|
|
|
|
|
def normalize_choice(value: str | None, choices: list[str], default: str) -> str:
|
|
value = (value or "").strip()
|
|
return value if value in choices else default
|
|
|
|
|
|
def make_case_code(db: Session, tenant_id: int, case_id: int, department: str) -> str:
|
|
prefix = "CASE"
|
|
dept = (department or "GEN").upper().replace(" ", "")[:4]
|
|
return f"{prefix}-{tenant_id}-{dept}-{case_id:06d}"
|
|
|
|
|
|
def list_clients_for_case(db: Session, *, tenant_id: int, branch_id: int | None) -> list[Client]:
|
|
stmt = select(Client).where(Client.tenant_id == tenant_id, Client.is_archived.is_(False))
|
|
if branch_id:
|
|
stmt = stmt.where(Client.branch_id == branch_id)
|
|
return list(db.execute(stmt.order_by(Client.client_name.asc())).scalars())
|
|
|
|
|
|
def list_engagements_for_client(
|
|
db: Session,
|
|
*,
|
|
tenant_id: int,
|
|
client_id: int,
|
|
financial_year: str | None = None,
|
|
) -> list[ClientServiceSubscription]:
|
|
stmt = select(ClientServiceSubscription).where(
|
|
ClientServiceSubscription.tenant_id == tenant_id,
|
|
ClientServiceSubscription.client_id == client_id,
|
|
)
|
|
if financial_year:
|
|
stmt = stmt.where(ClientServiceSubscription.financial_year == financial_year.strip())
|
|
return list(
|
|
db.execute(
|
|
stmt.order_by(ClientServiceSubscription.financial_year.desc(), ClientServiceSubscription.id.desc())
|
|
).scalars()
|
|
)
|
|
|
|
|
|
def list_assignable_users(db: Session, *, tenant_id: int, branch_id: int | None):
|
|
from app.modules.core.iam.models import User
|
|
|
|
stmt = select(User).where(User.tenant_id == tenant_id)
|
|
if branch_id:
|
|
stmt = stmt.where(or_(User.branch_id == branch_id, User.branch_id.is_(None)))
|
|
order_cols = [User.full_name.asc()]
|
|
if hasattr(User, "login_id"):
|
|
order_cols.append(User.login_id.asc())
|
|
elif hasattr(User, "email"):
|
|
order_cols.append(User.email.asc())
|
|
return list(db.execute(stmt.order_by(*order_cols)).scalars())
|
|
|
|
|
|
def list_cases(
|
|
db: Session,
|
|
*,
|
|
tenant_id: int,
|
|
branch_id: int | None,
|
|
q: str = "",
|
|
department: str = "",
|
|
status: str = "",
|
|
include_archived: bool = False,
|
|
financial_year: str | None = None,
|
|
assessment_year: str | None = None,
|
|
) -> list[NoticeCase]:
|
|
stmt = (
|
|
select(NoticeCase)
|
|
.options(joinedload(NoticeCase.client), joinedload(NoticeCase.assigned_partner), joinedload(NoticeCase.assigned_manager), joinedload(NoticeCase.assigned_staff))
|
|
.where(NoticeCase.tenant_id == tenant_id)
|
|
)
|
|
if branch_id:
|
|
stmt = stmt.where(NoticeCase.branch_id == branch_id)
|
|
if not include_archived:
|
|
stmt = stmt.where(NoticeCase.is_archived.is_(False))
|
|
if department:
|
|
stmt = stmt.where(NoticeCase.department == department)
|
|
if status:
|
|
stmt = stmt.where(NoticeCase.status == status)
|
|
fy = (financial_year or "").strip()
|
|
ay = (assessment_year or "").strip()
|
|
if fy:
|
|
stmt = stmt.where(NoticeCase.financial_year == fy)
|
|
elif ay:
|
|
stmt = stmt.where(NoticeCase.assessment_year == ay)
|
|
q = (q or "").strip()
|
|
if q:
|
|
like = f"%{q}%"
|
|
stmt = stmt.join(Client, Client.id == NoticeCase.client_id).where(
|
|
or_(
|
|
NoticeCase.case_code.ilike(like),
|
|
NoticeCase.title.ilike(like),
|
|
NoticeCase.reference_no.ilike(like),
|
|
NoticeCase.din_ack_no.ilike(like),
|
|
Client.client_name.ilike(like),
|
|
Client.client_code.ilike(like),
|
|
Client.pan.ilike(like),
|
|
Client.gstin.ilike(like),
|
|
)
|
|
)
|
|
return list(db.execute(stmt.order_by(NoticeCase.due_date.asc().nullslast(), NoticeCase.updated_at_utc.desc())).unique().scalars())
|
|
|
|
|
|
def case_dashboard_summary(rows: list[NoticeCase]) -> dict[str, int]:
|
|
today = date.today()
|
|
open_statuses = {"open", "reply_pending", "hearing_scheduled", "appeal_pending"}
|
|
summary = {
|
|
"total": len(rows),
|
|
"open": 0,
|
|
"reply_pending": 0,
|
|
"hearing_scheduled": 0,
|
|
"overdue": 0,
|
|
"closed": 0,
|
|
}
|
|
for row in rows:
|
|
status = (row.status or "").strip().lower()
|
|
if status in open_statuses:
|
|
summary["open"] += 1
|
|
if status == "reply_pending":
|
|
summary["reply_pending"] += 1
|
|
if status == "hearing_scheduled":
|
|
summary["hearing_scheduled"] += 1
|
|
if status in {"closed", "archived"}:
|
|
summary["closed"] += 1
|
|
if row.due_date and row.due_date < today and status not in {"closed", "archived"}:
|
|
summary["overdue"] += 1
|
|
return summary
|
|
|
|
|
|
def get_case(db: Session, case_id: int) -> NoticeCase | None:
|
|
return db.execute(
|
|
select(NoticeCase)
|
|
.options(
|
|
joinedload(NoticeCase.client),
|
|
joinedload(NoticeCase.engagement),
|
|
joinedload(NoticeCase.assigned_partner),
|
|
joinedload(NoticeCase.assigned_manager),
|
|
joinedload(NoticeCase.assigned_staff),
|
|
joinedload(NoticeCase.events),
|
|
joinedload(NoticeCase.hearings),
|
|
joinedload(NoticeCase.orders),
|
|
joinedload(NoticeCase.documents).joinedload(NoticeCaseDocument.uploaded_by),
|
|
)
|
|
.where(NoticeCase.id == case_id)
|
|
).unique().scalar_one_or_none()
|
|
|
|
|
|
def create_case(db: Session, *, tenant_id: int, branch_id: int | None, user, data: dict) -> NoticeCase:
|
|
client = db.get(Client, int(data["client_id"]))
|
|
if not client or client.tenant_id != tenant_id:
|
|
raise ValueError("Invalid client selected.")
|
|
if branch_id and client.branch_id != branch_id:
|
|
raise ValueError("Selected client does not belong to the active branch.")
|
|
engagement_id = int(data["engagement_id"]) if data.get("engagement_id") else None
|
|
active_fy = (data.get("active_financial_year") or "").strip()[:9] or None
|
|
active_ay = (data.get("active_assessment_year") or "").strip()[:9] or None
|
|
selected_fy = (data.get("financial_year") or "").strip()[:9] or active_fy
|
|
selected_ay = (data.get("assessment_year") or "").strip()[:9] or active_ay
|
|
if engagement_id:
|
|
engagement = db.get(ClientServiceSubscription, engagement_id)
|
|
if not engagement or engagement.tenant_id != tenant_id or engagement.client_id != client.id:
|
|
raise ValueError("Invalid engagement selected.")
|
|
engagement_fy = (engagement.financial_year or "").strip()
|
|
engagement_ay = (engagement.assessment_year or "").strip()
|
|
if selected_fy and engagement_fy and selected_fy != engagement_fy:
|
|
raise ValueError("Selected engagement belongs to a different financial year.")
|
|
selected_fy = selected_fy or engagement_fy or None
|
|
selected_ay = selected_ay or engagement_ay or None
|
|
row = NoticeCase(
|
|
tenant_id=tenant_id,
|
|
branch_id=client.branch_id,
|
|
client_id=client.id,
|
|
engagement_id=engagement_id,
|
|
case_code="PENDING",
|
|
department=normalize_choice(data.get("department"), DEPARTMENTS, "GST"),
|
|
case_type=normalize_choice(data.get("case_type"), CASE_TYPES, "Notice"),
|
|
title=(data.get("title") or "").strip()[:255],
|
|
reference_no=(data.get("reference_no") or "").strip()[:150] or None,
|
|
din_ack_no=(data.get("din_ack_no") or "").strip()[:150] or None,
|
|
notice_date=parse_date(data.get("notice_date")),
|
|
due_date=parse_date(data.get("due_date")),
|
|
financial_year=selected_fy,
|
|
assessment_year=selected_ay,
|
|
period_label=(data.get("period_label") or "").strip()[:60] or None,
|
|
status=normalize_choice(data.get("status"), CASE_STATUSES, "open"),
|
|
priority=normalize_choice(data.get("priority"), CASE_PRIORITIES, "normal"),
|
|
issue_summary=(data.get("issue_summary") or "").strip() or None,
|
|
remarks=(data.get("remarks") or "").strip() or None,
|
|
assigned_partner_user_id=int(data["assigned_partner_user_id"]) if data.get("assigned_partner_user_id") else None,
|
|
assigned_manager_user_id=int(data["assigned_manager_user_id"]) if data.get("assigned_manager_user_id") else None,
|
|
assigned_staff_user_id=int(data["assigned_staff_user_id"]) if data.get("assigned_staff_user_id") else None,
|
|
created_by_user_id=user.id,
|
|
updated_by_user_id=user.id,
|
|
)
|
|
if not row.title:
|
|
raise ValueError("Case title is required.")
|
|
db.add(row)
|
|
db.flush()
|
|
row.case_code = make_case_code(db, tenant_id, row.id, row.department)
|
|
return row
|
|
|
|
|
|
def update_case(db: Session, *, case: NoticeCase, user, data: dict) -> NoticeCase:
|
|
case.department = normalize_choice(data.get("department"), DEPARTMENTS, case.department)
|
|
case.case_type = normalize_choice(data.get("case_type"), CASE_TYPES, case.case_type)
|
|
case.title = (data.get("title") or case.title).strip()[:255]
|
|
case.reference_no = (data.get("reference_no") or "").strip()[:150] or None
|
|
case.din_ack_no = (data.get("din_ack_no") or "").strip()[:150] or None
|
|
case.notice_date = parse_date(data.get("notice_date"))
|
|
case.due_date = parse_date(data.get("due_date"))
|
|
active_fy = (data.get("active_financial_year") or "").strip()[:9] or None
|
|
active_ay = (data.get("active_assessment_year") or "").strip()[:9] or None
|
|
case.financial_year = (data.get("financial_year") or "").strip()[:9] or active_fy
|
|
case.assessment_year = (data.get("assessment_year") or "").strip()[:9] or active_ay
|
|
case.period_label = (data.get("period_label") or "").strip()[:60] or None
|
|
case.status = normalize_choice(data.get("status"), CASE_STATUSES, case.status)
|
|
case.priority = normalize_choice(data.get("priority"), CASE_PRIORITIES, case.priority)
|
|
case.issue_summary = (data.get("issue_summary") or "").strip() or None
|
|
case.remarks = (data.get("remarks") or "").strip() or None
|
|
case.assigned_partner_user_id = int(data["assigned_partner_user_id"]) if data.get("assigned_partner_user_id") else None
|
|
case.assigned_manager_user_id = int(data["assigned_manager_user_id"]) if data.get("assigned_manager_user_id") else None
|
|
case.assigned_staff_user_id = int(data["assigned_staff_user_id"]) if data.get("assigned_staff_user_id") else None
|
|
case.updated_by_user_id = user.id
|
|
return case
|
|
|
|
|
|
def add_event(db: Session, *, case: NoticeCase, user, data: dict) -> NoticeCaseEvent:
|
|
row = NoticeCaseEvent(
|
|
tenant_id=case.tenant_id,
|
|
branch_id=case.branch_id,
|
|
case_id=case.id,
|
|
event_type=normalize_choice(data.get("event_type"), EVENT_TYPES, "Internal Note"),
|
|
event_date=parse_date(data.get("event_date")) or date.today(),
|
|
description=(data.get("description") or "").strip(),
|
|
next_due_date=parse_date(data.get("next_due_date")),
|
|
created_by_user_id=user.id,
|
|
)
|
|
if not row.description:
|
|
raise ValueError("Event description is required.")
|
|
if row.next_due_date:
|
|
case.due_date = row.next_due_date
|
|
case.updated_by_user_id = user.id
|
|
db.add(row)
|
|
return row
|
|
|
|
|
|
def add_hearing(db: Session, *, case: NoticeCase, user, data: dict) -> NoticeCaseHearing:
|
|
row = NoticeCaseHearing(
|
|
tenant_id=case.tenant_id,
|
|
branch_id=case.branch_id,
|
|
case_id=case.id,
|
|
hearing_date=parse_date(data.get("hearing_date")) or date.today(),
|
|
hearing_time=(data.get("hearing_time") or "").strip()[:20] or None,
|
|
venue_or_mode=(data.get("venue_or_mode") or "").strip()[:200] or None,
|
|
officer_name=(data.get("officer_name") or "").strip()[:150] or None,
|
|
agenda=(data.get("agenda") or "").strip() or None,
|
|
outcome=(data.get("outcome") or "").strip() or None,
|
|
status=normalize_choice(data.get("status"), HEARING_STATUSES, "scheduled"),
|
|
created_by_user_id=user.id,
|
|
)
|
|
case.status = "hearing_scheduled" if row.status == "scheduled" else case.status
|
|
case.due_date = row.hearing_date
|
|
case.updated_by_user_id = user.id
|
|
db.add(row)
|
|
return row
|
|
|
|
|
|
def add_order(db: Session, *, case: NoticeCase, user, data: dict) -> NoticeCaseOrder:
|
|
row = NoticeCaseOrder(
|
|
tenant_id=case.tenant_id,
|
|
branch_id=case.branch_id,
|
|
case_id=case.id,
|
|
order_type=normalize_choice(data.get("order_type"), ORDER_TYPES, "Other"),
|
|
order_no=(data.get("order_no") or "").strip()[:150] or None,
|
|
order_date=parse_date(data.get("order_date")) or date.today(),
|
|
demand_amount=int(data.get("demand_amount") or 0),
|
|
tax_amount=int(data.get("tax_amount") or 0),
|
|
interest_amount=int(data.get("interest_amount") or 0),
|
|
penalty_amount=int(data.get("penalty_amount") or 0),
|
|
summary=(data.get("summary") or "").strip() or None,
|
|
appeal_due_date=parse_date(data.get("appeal_due_date")),
|
|
appeal_filed=bool(data.get("appeal_filed")),
|
|
created_by_user_id=user.id,
|
|
)
|
|
case.status = "appeal_filed" if row.appeal_filed else "order_received"
|
|
case.due_date = row.appeal_due_date
|
|
case.updated_by_user_id = user.id
|
|
db.add(row)
|
|
return row
|
|
|
|
|
|
class UploadValidationError(ValueError):
|
|
"""Raised when a notice/case document upload violates branch file policy."""
|
|
|
|
|
|
DANGEROUS_DOCUMENT_EXTENSIONS = {
|
|
".exe", ".bat", ".cmd", ".com", ".scr", ".pif",
|
|
".sh", ".bash", ".zsh", ".ps1", ".psm1",
|
|
".js", ".jse", ".vbs", ".vbe", ".wsf",
|
|
".msi", ".dll", ".jar", ".apk",
|
|
}
|
|
|
|
|
|
def _branch_document_settings(db: Session, branch_id: int | None) -> BranchSettings | None:
|
|
if not branch_id:
|
|
return None
|
|
return db.execute(
|
|
select(BranchSettings).where(BranchSettings.branch_id == int(branch_id))
|
|
).scalar_one_or_none()
|
|
|
|
|
|
def _normalise_allowed_extensions(raw: str | None) -> set[str]:
|
|
values = raw or "pdf,jpg,jpeg,png,xlsx,xls,docx,zip"
|
|
return {
|
|
"." + item.strip().lower().lstrip(".")
|
|
for item in values.split(",")
|
|
if item.strip()
|
|
}
|
|
|
|
|
|
def notice_case_upload_policy(db: Session, *, branch_id: int | None) -> dict[str, object]:
|
|
settings = _branch_document_settings(db, branch_id)
|
|
allowed_exts = _normalise_allowed_extensions(getattr(settings, "allowed_ext_csv", None))
|
|
max_file_mb = int(getattr(settings, "max_file_mb", 25) or 25)
|
|
if max_file_mb <= 0:
|
|
max_file_mb = 25
|
|
return {
|
|
"allowed_exts": sorted(allowed_exts),
|
|
"allowed_ext_csv": ",".join(ext.lstrip(".") for ext in sorted(allowed_exts)),
|
|
"accept": ",".join(sorted(allowed_exts)),
|
|
"max_file_mb": max_file_mb,
|
|
"max_file_bytes": max_file_mb * 1024 * 1024,
|
|
}
|
|
|
|
|
|
def _validate_notice_case_upload_policy(db: Session, *, case: NoticeCase, upload_file: UploadFile, original: str) -> bytes:
|
|
policy = notice_case_upload_policy(db, branch_id=case.branch_id)
|
|
allowed_exts = set(policy["allowed_exts"] or [])
|
|
max_file_bytes = int(policy["max_file_bytes"] or (25 * 1024 * 1024))
|
|
|
|
if not original or original in {".", ".."}:
|
|
raise UploadValidationError("Invalid filename.")
|
|
|
|
ext = Path(original).suffix.lower()
|
|
if not ext:
|
|
raise UploadValidationError("File extension is required.")
|
|
|
|
if ext in DANGEROUS_DOCUMENT_EXTENSIONS:
|
|
raise UploadValidationError("Executable or dangerous file types are not allowed.")
|
|
|
|
if allowed_exts and ext not in allowed_exts:
|
|
raise UploadValidationError("File type not allowed. Allowed types: " + ", ".join(sorted(allowed_exts)))
|
|
|
|
data = upload_file.file.read(max_file_bytes + 1)
|
|
if len(data) > max_file_bytes:
|
|
raise UploadValidationError(f"File too large. Maximum allowed size is {policy['max_file_mb']} MB.")
|
|
|
|
return data
|
|
|
|
|
|
def _storage_root() -> Path:
|
|
settings = get_settings()
|
|
base = getattr(settings, "LOCAL_STORAGE_ROOT", None) or getattr(settings, "DOCUMENT_STORAGE_ROOT", None) or "data/storage"
|
|
return Path(base)
|
|
|
|
|
|
def case_document_absolute_path(document: NoticeCaseDocument) -> Path:
|
|
return _storage_root() / document.local_relative_path
|
|
|
|
|
|
def save_case_document(db: Session, *, case: NoticeCase, upload_file: UploadFile, user, title: str, document_type: str, description: str | None, event_id: int | None = None) -> NoticeCaseDocument:
|
|
original = Path(upload_file.filename or "").name
|
|
data = _validate_notice_case_upload_policy(db, case=case, upload_file=upload_file, original=original)
|
|
ext = Path(original).suffix.lower()
|
|
stored = f"case_{case.id}_{uuid4().hex}{ext}"
|
|
fy_folder = f"FY{case.financial_year}" if case.financial_year else "FY_UNASSIGNED"
|
|
rel = Path("notice_cases") / fy_folder / str(case.tenant_id) / str(case.client_id) / case.case_code / stored
|
|
absolute = _storage_root() / rel
|
|
absolute.parent.mkdir(parents=True, exist_ok=True)
|
|
absolute.write_bytes(data)
|
|
latest_version = db.execute(select(func.max(NoticeCaseDocument.version_no)).where(NoticeCaseDocument.case_id == case.id, NoticeCaseDocument.title == (title or original))).scalar_one() or 0
|
|
row = NoticeCaseDocument(
|
|
tenant_id=case.tenant_id,
|
|
branch_id=case.branch_id,
|
|
case_id=case.id,
|
|
event_id=event_id,
|
|
document_type=normalize_choice(document_type, CASE_DOCUMENT_TYPES, "OTHER"),
|
|
title=(title or original).strip()[:255],
|
|
description=(description or "").strip() or None,
|
|
original_filename=original,
|
|
stored_filename=stored,
|
|
content_type=upload_file.content_type,
|
|
file_size_bytes=len(data),
|
|
local_relative_path=str(rel).replace("\\", "/"),
|
|
version_no=int(latest_version) + 1,
|
|
uploaded_by_user_id=user.id,
|
|
)
|
|
db.add(row)
|
|
case.updated_by_user_id = user.id
|
|
return row
|