feat: reimport pdf
This commit is contained in:
@@ -3,13 +3,15 @@
|
||||
import json
|
||||
from datetime import datetime
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
from fastapi import APIRouter, Depends, HTTPException, File, UploadFile, Form
|
||||
from fastapi.responses import Response, JSONResponse
|
||||
from sqlalchemy.orm import Session
|
||||
from sqlalchemy import func, select
|
||||
|
||||
from ...database import get_session, DatabaseService
|
||||
from ...database import get_session, DatabaseService, storage
|
||||
from ...database.models import Document, Immeuble, Lot, Locataire, Revenu, Depense
|
||||
from ...database.service import DuplicateDocumentError
|
||||
from ...extractor import extract_compte_rendu
|
||||
from ..schemas import SaveRequest, SaveResponse, DocumentSummary
|
||||
|
||||
router = APIRouter(prefix="/api", tags=["documents"])
|
||||
@@ -64,6 +66,80 @@ async def save_document(
|
||||
)
|
||||
|
||||
|
||||
@router.post("/save-with-pdf", response_model=SaveResponse)
|
||||
async def save_document_with_pdf(
|
||||
pdf_file: UploadFile = File(..., description="Fichier PDF original"),
|
||||
data: str = Form(..., description="Donnees JSON extraites"),
|
||||
depenses_tags: str | None = Form(None, description="Tags JSON pour les depenses"),
|
||||
overwrite: bool = Form(False, description="Ecraser si existant"),
|
||||
session: Session = Depends(get_session),
|
||||
) -> SaveResponse:
|
||||
"""Sauvegarde les donnees extraites avec le fichier PDF original.
|
||||
|
||||
Cette version stocke le PDF et le JSON sur le disque pour tracabilite.
|
||||
|
||||
- **pdf_file**: Fichier PDF original (multipart)
|
||||
- **data**: Donnees JSON extraites (string JSON)
|
||||
- **depenses_tags**: Tags JSON pour les depenses (optionnel)
|
||||
- **overwrite**: Si True, ecrase le document existant
|
||||
"""
|
||||
# Parse JSON data
|
||||
try:
|
||||
parsed_data = json.loads(data)
|
||||
except json.JSONDecodeError as e:
|
||||
raise HTTPException(status_code=400, detail=f"JSON invalide pour data: {e}")
|
||||
|
||||
# Parse depenses_tags if provided
|
||||
parsed_tags = None
|
||||
if depenses_tags:
|
||||
try:
|
||||
parsed_tags = json.loads(depenses_tags)
|
||||
except json.JSONDecodeError as e:
|
||||
raise HTTPException(
|
||||
status_code=400, detail=f"JSON invalide pour depenses_tags: {e}"
|
||||
)
|
||||
|
||||
# Read PDF content
|
||||
try:
|
||||
pdf_content = await pdf_file.read()
|
||||
except Exception as e:
|
||||
raise HTTPException(status_code=400, detail=f"Erreur lecture du PDF: {str(e)}")
|
||||
|
||||
try:
|
||||
db_service = DatabaseService(session)
|
||||
document = db_service.save_document(
|
||||
data=parsed_data,
|
||||
source_file=pdf_file.filename,
|
||||
pdf_content=pdf_content,
|
||||
depenses_tags=parsed_tags,
|
||||
overwrite=overwrite,
|
||||
)
|
||||
|
||||
return SaveResponse(
|
||||
success=True,
|
||||
message="Document sauvegarde avec succes (PDF et JSON stockes)",
|
||||
document_id=document.id,
|
||||
reference=document.reference,
|
||||
date=str(document.date),
|
||||
)
|
||||
|
||||
except DuplicateDocumentError as e:
|
||||
return SaveResponse(
|
||||
success=False,
|
||||
message=f"Document deja existant: reference={e.reference}, date={e.date}",
|
||||
reference=e.reference,
|
||||
date=str(e.date),
|
||||
)
|
||||
|
||||
except ValueError as e:
|
||||
raise HTTPException(status_code=400, detail=str(e))
|
||||
|
||||
except Exception as e:
|
||||
raise HTTPException(
|
||||
status_code=500, detail=f"Erreur lors de la sauvegarde: {str(e)}"
|
||||
)
|
||||
|
||||
|
||||
@router.get("/stats")
|
||||
async def get_stats(
|
||||
session: Session = Depends(get_session),
|
||||
@@ -110,6 +186,8 @@ async def list_documents(
|
||||
solde_montant=doc.solde_montant,
|
||||
solde_type=doc.solde_type,
|
||||
created_at=doc.created_at.isoformat() if doc.created_at else None,
|
||||
has_pdf=doc.pdf_path is not None,
|
||||
has_json=doc.json_path is not None,
|
||||
)
|
||||
for doc in documents
|
||||
]
|
||||
@@ -157,6 +235,8 @@ async def get_document(
|
||||
},
|
||||
"json_data": json.loads(document.json_data) if document.json_data else None,
|
||||
"created_at": document.created_at.isoformat() if document.created_at else None,
|
||||
"has_pdf": document.pdf_path is not None,
|
||||
"has_json": document.json_path is not None,
|
||||
}
|
||||
|
||||
|
||||
@@ -189,3 +269,135 @@ async def check_duplicate(
|
||||
"reference": reference,
|
||||
"date": date,
|
||||
}
|
||||
|
||||
|
||||
@router.get("/documents/{document_id}/pdf")
|
||||
async def download_document_pdf(
|
||||
document_id: int,
|
||||
session: Session = Depends(get_session),
|
||||
) -> Response:
|
||||
"""Telecharge le fichier PDF original d'un document.
|
||||
|
||||
Retourne le fichier PDF si disponible, sinon erreur 404.
|
||||
"""
|
||||
db_service = DatabaseService(session)
|
||||
document = db_service.get_document_by_id(document_id)
|
||||
|
||||
if not document:
|
||||
raise HTTPException(status_code=404, detail="Document non trouve")
|
||||
|
||||
if not document.pdf_path:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Fichier PDF non disponible pour ce document",
|
||||
)
|
||||
|
||||
try:
|
||||
pdf_content = storage.read_pdf(document.pdf_path)
|
||||
filename = document.source_file or f"{document.reference}.pdf"
|
||||
return Response(
|
||||
content=pdf_content,
|
||||
media_type="application/pdf",
|
||||
headers={"Content-Disposition": f'inline; filename="{filename}"'},
|
||||
)
|
||||
except FileNotFoundError:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Fichier PDF introuvable sur le disque",
|
||||
)
|
||||
|
||||
|
||||
@router.get("/documents/{document_id}/json")
|
||||
async def download_document_json(
|
||||
document_id: int,
|
||||
session: Session = Depends(get_session),
|
||||
) -> Response:
|
||||
"""Telecharge le fichier JSON extrait d'un document.
|
||||
|
||||
Retourne le fichier JSON si disponible sur disque,
|
||||
sinon retourne le json_data de la base de donnees.
|
||||
"""
|
||||
db_service = DatabaseService(session)
|
||||
document = db_service.get_document_by_id(document_id)
|
||||
|
||||
if not document:
|
||||
raise HTTPException(status_code=404, detail="Document non trouve")
|
||||
|
||||
filename = f"{document.reference}_{document.date}.json"
|
||||
|
||||
# Try to read from storage first
|
||||
if document.json_path:
|
||||
try:
|
||||
json_data = storage.read_json(document.json_path)
|
||||
return JSONResponse(
|
||||
content=json_data,
|
||||
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
|
||||
)
|
||||
except FileNotFoundError:
|
||||
pass # Fall back to database
|
||||
|
||||
# Fall back to json_data in database
|
||||
if document.json_data:
|
||||
json_data = json.loads(document.json_data)
|
||||
return JSONResponse(
|
||||
content=json_data,
|
||||
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
|
||||
)
|
||||
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Donnees JSON non disponibles pour ce document",
|
||||
)
|
||||
|
||||
|
||||
@router.post("/documents/{document_id}/re-extract")
|
||||
async def re_extract_document(
|
||||
document_id: int,
|
||||
session: Session = Depends(get_session),
|
||||
) -> dict:
|
||||
"""Re-extrait les donnees depuis le PDF stocke.
|
||||
|
||||
Utile pour corriger l'extraction apres amelioration des parsers.
|
||||
Ne modifie pas automatiquement la base - retourne les nouvelles donnees
|
||||
pour validation par l'utilisateur.
|
||||
|
||||
Retourne les donnees re-extraites du PDF.
|
||||
"""
|
||||
db_service = DatabaseService(session)
|
||||
document = db_service.get_document_by_id(document_id)
|
||||
|
||||
if not document:
|
||||
raise HTTPException(status_code=404, detail="Document non trouve")
|
||||
|
||||
if not document.pdf_path:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Fichier PDF non disponible pour ce document - re-extraction impossible",
|
||||
)
|
||||
|
||||
# Get absolute path
|
||||
pdf_full_path = storage.get_absolute_path(document.pdf_path)
|
||||
|
||||
if not pdf_full_path.exists():
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Fichier PDF introuvable sur le disque",
|
||||
)
|
||||
|
||||
# Re-extract
|
||||
try:
|
||||
new_data = extract_compte_rendu(str(pdf_full_path))
|
||||
except Exception as e:
|
||||
raise HTTPException(
|
||||
status_code=422,
|
||||
detail=f"Erreur lors de la re-extraction: {str(e)}",
|
||||
)
|
||||
|
||||
return {
|
||||
"document_id": document_id,
|
||||
"reference": document.reference,
|
||||
"date": str(document.date),
|
||||
"original_json_path": document.json_path,
|
||||
"re_extracted_data": new_data,
|
||||
"message": "Donnees re-extraites. Utilisez PUT /api/documents/{id} pour mettre a jour.",
|
||||
}
|
||||
|
||||
@@ -54,6 +54,8 @@ class DocumentSummary(BaseModel):
|
||||
solde_montant: float | None
|
||||
solde_type: str | None
|
||||
created_at: str
|
||||
has_pdf: bool = False
|
||||
has_json: bool = False
|
||||
|
||||
|
||||
# ============================================================
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
from .connection import get_engine, get_session, get_session_factory, init_db
|
||||
from .models import Base, Document, Immeuble, Lot, Locataire, Revenu, Depense
|
||||
from .service import DatabaseService, DuplicateDocumentError
|
||||
from . import storage
|
||||
|
||||
__all__ = [
|
||||
"get_engine",
|
||||
@@ -18,4 +19,5 @@ __all__ = [
|
||||
"Depense",
|
||||
"DatabaseService",
|
||||
"DuplicateDocumentError",
|
||||
"storage",
|
||||
]
|
||||
|
||||
@@ -144,6 +144,14 @@ class Document(Base):
|
||||
|
||||
created_at = Column(DateTime, default=datetime.utcnow)
|
||||
|
||||
# Chemins vers les fichiers stockés (relatifs à PLESNA_STORAGE_PATH)
|
||||
pdf_path = Column(
|
||||
String(500), nullable=True
|
||||
) # ex: "2024/S_REF-01234_2024-01-15.pdf"
|
||||
json_path = Column(
|
||||
String(500), nullable=True
|
||||
) # ex: "2024/S_REF-01234_2024-01-15.json"
|
||||
|
||||
# Clé unique: référence + date (évite les doublons)
|
||||
__table_args__ = (
|
||||
UniqueConstraint("reference", "date", name="uq_document_reference_date"),
|
||||
|
||||
@@ -9,6 +9,7 @@ from sqlalchemy.orm import Session
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
|
||||
from .models import Document, Immeuble, Lot, Locataire, Revenu, Depense, Tag
|
||||
from . import storage
|
||||
|
||||
|
||||
class DuplicateDocumentError(Exception):
|
||||
@@ -98,6 +99,7 @@ class DatabaseService:
|
||||
self,
|
||||
data: dict[str, Any],
|
||||
source_file: str = None,
|
||||
pdf_content: bytes = None,
|
||||
depenses_tags: list[dict] = None,
|
||||
overwrite: bool = False,
|
||||
) -> Document:
|
||||
@@ -107,6 +109,7 @@ class DatabaseService:
|
||||
data: The 'data' portion of the extracted JSON (contains metadata,
|
||||
situation_locataires, recapitulatif_operations)
|
||||
source_file: Original PDF filename
|
||||
pdf_content: Binary content of the PDF file (optional, for storage)
|
||||
depenses_tags: List of tags for expenses
|
||||
overwrite: If True, delete existing document and recreate it
|
||||
|
||||
@@ -129,11 +132,17 @@ class DatabaseService:
|
||||
if not reference or not doc_date:
|
||||
raise ValueError("Document must have reference and date")
|
||||
|
||||
# Check for duplicates
|
||||
# Check for duplicates and preserve existing file paths if overwriting
|
||||
existing_pdf_path = None
|
||||
existing_json_path = None
|
||||
existing = self.check_duplicate(reference, doc_date)
|
||||
if existing:
|
||||
if overwrite:
|
||||
# Preserve existing file paths for reuse
|
||||
existing_pdf_path = existing.pdf_path
|
||||
existing_json_path = existing.json_path
|
||||
# Delete existing document (cascade will delete related data)
|
||||
# But DON'T delete files - we'll reuse or update them
|
||||
self.session.delete(existing)
|
||||
self.session.flush()
|
||||
else:
|
||||
@@ -147,6 +156,26 @@ class DatabaseService:
|
||||
code_postal=immeuble_info.get("code_postal"),
|
||||
)
|
||||
|
||||
# Compute storage paths for PDF and JSON
|
||||
pdf_path = None
|
||||
json_path = None
|
||||
if pdf_content is not None:
|
||||
# New PDF provided - compute new paths
|
||||
pdf_path, json_path = storage.compute_document_paths(
|
||||
reference=reference,
|
||||
doc_date=doc_date,
|
||||
immeuble_adresse=immeuble.adresse,
|
||||
)
|
||||
# Delete old files if paths are different
|
||||
if existing_pdf_path and existing_pdf_path != pdf_path:
|
||||
storage.delete_document_files(existing_pdf_path, None)
|
||||
if existing_json_path and existing_json_path != json_path:
|
||||
storage.delete_document_files(None, existing_json_path)
|
||||
elif existing_json_path:
|
||||
# No new PDF but we have existing paths - preserve them
|
||||
pdf_path = existing_pdf_path
|
||||
json_path = existing_json_path
|
||||
|
||||
# Create document
|
||||
document = Document(
|
||||
reference=reference,
|
||||
@@ -160,10 +189,21 @@ class DatabaseService:
|
||||
solde_montant=solde_info.get("montant"),
|
||||
solde_type=solde_info.get("type"),
|
||||
solde_date_arrete=self._parse_date(solde_info.get("date_arrete")),
|
||||
pdf_path=pdf_path,
|
||||
json_path=json_path,
|
||||
)
|
||||
self.session.add(document)
|
||||
self.session.flush()
|
||||
|
||||
# Save files to storage
|
||||
if pdf_content is not None and pdf_path and json_path:
|
||||
# New PDF provided - save both files
|
||||
storage.save_pdf(pdf_content, pdf_path)
|
||||
storage.save_json(data, json_path)
|
||||
elif json_path:
|
||||
# No new PDF but we have a json_path - update the JSON file
|
||||
storage.save_json(data, json_path)
|
||||
|
||||
# Process situation_locataires (revenus)
|
||||
for situation in data.get("situation_locataires", []):
|
||||
self._save_situation_locataire(document, immeuble, situation)
|
||||
|
||||
280
src/plesna_gerance/database/storage.py
Normal file
280
src/plesna_gerance/database/storage.py
Normal file
@@ -0,0 +1,280 @@
|
||||
"""Service de stockage des fichiers PDF et JSON."""
|
||||
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
from datetime import date
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
def _get_project_root() -> Path:
|
||||
"""Find project root by looking for pyproject.toml."""
|
||||
current = Path(__file__).resolve()
|
||||
for parent in current.parents:
|
||||
if (parent / "pyproject.toml").exists():
|
||||
return parent
|
||||
return Path.home() / ".plesna_gerance"
|
||||
|
||||
|
||||
def get_storage_root() -> Path:
|
||||
"""Get storage root from environment or default.
|
||||
|
||||
Returns:
|
||||
Path to the documents storage directory.
|
||||
"""
|
||||
env_path = os.environ.get("PLESNA_STORAGE_PATH")
|
||||
if env_path:
|
||||
return Path(env_path)
|
||||
return _get_project_root() / "data" / "documents"
|
||||
|
||||
|
||||
def extract_street_letter(adresse: str | None) -> str:
|
||||
"""Extract the first letter of the main street name.
|
||||
|
||||
Examples:
|
||||
"4 RUE SERVIENT" -> "S"
|
||||
"33 RUE MARC BLOCH" -> "M"
|
||||
"1 RUE MARIETTON" -> "M"
|
||||
"12 AVENUE JEAN JAURES" -> "J"
|
||||
None -> "X"
|
||||
|
||||
Args:
|
||||
adresse: The address string (e.g., "4 RUE SERVIENT")
|
||||
|
||||
Returns:
|
||||
Single uppercase letter representing the street, or "X" if unknown.
|
||||
"""
|
||||
if not adresse:
|
||||
return "X"
|
||||
|
||||
# Normalize: uppercase, remove extra spaces
|
||||
adresse = adresse.upper().strip()
|
||||
|
||||
# Remove common street type prefixes to get to the street name
|
||||
# Pattern: [number] [street type] [street name]
|
||||
street_types = [
|
||||
r"^\d+\s+", # Leading number
|
||||
r"^RUE\s+",
|
||||
r"^AVENUE\s+",
|
||||
r"^BOULEVARD\s+",
|
||||
r"^PLACE\s+",
|
||||
r"^ALLEE\s+",
|
||||
r"^IMPASSE\s+",
|
||||
r"^CHEMIN\s+",
|
||||
r"^COURS\s+",
|
||||
r"^PASSAGE\s+",
|
||||
r"^QUAI\s+",
|
||||
]
|
||||
|
||||
remaining = adresse
|
||||
for pattern in street_types:
|
||||
remaining = re.sub(pattern, "", remaining, flags=re.IGNORECASE)
|
||||
|
||||
# Get first letter of what remains
|
||||
remaining = remaining.strip()
|
||||
if remaining and remaining[0].isalpha():
|
||||
return remaining[0].upper()
|
||||
|
||||
return "X"
|
||||
|
||||
|
||||
def sanitize_filename(name: str) -> str:
|
||||
"""Sanitize a string to be used as filename.
|
||||
|
||||
Args:
|
||||
name: The string to sanitize.
|
||||
|
||||
Returns:
|
||||
Sanitized string safe for filenames.
|
||||
"""
|
||||
# Replace problematic characters
|
||||
sanitized = re.sub(r'[<>:"/\\|?*]', "_", name)
|
||||
# Remove leading/trailing spaces and dots
|
||||
sanitized = sanitized.strip(". ")
|
||||
return sanitized
|
||||
|
||||
|
||||
def compute_document_paths(
|
||||
reference: str,
|
||||
doc_date: date,
|
||||
immeuble_adresse: str | None = None,
|
||||
) -> tuple[str, str]:
|
||||
"""Compute the relative paths for PDF and JSON storage.
|
||||
|
||||
Naming convention: {YYYY}/{L}_{reference}_{YYYY-MM-DD}.{ext}
|
||||
Where L is the first letter of the street name.
|
||||
|
||||
Args:
|
||||
reference: Document reference (e.g., "01234")
|
||||
doc_date: Document date
|
||||
immeuble_adresse: Address to extract street letter from
|
||||
|
||||
Returns:
|
||||
Tuple of (pdf_relative_path, json_relative_path)
|
||||
"""
|
||||
year = str(doc_date.year)
|
||||
letter = extract_street_letter(immeuble_adresse)
|
||||
date_str = doc_date.strftime("%Y-%m-%d")
|
||||
|
||||
# Sanitize reference for filename
|
||||
safe_ref = sanitize_filename(reference)
|
||||
|
||||
# Build filename: L_reference_date
|
||||
base_name = f"{letter}_{safe_ref}_{date_str}"
|
||||
|
||||
pdf_path = f"{year}/{base_name}.pdf"
|
||||
json_path = f"{year}/{base_name}.json"
|
||||
|
||||
return pdf_path, json_path
|
||||
|
||||
|
||||
def save_pdf(
|
||||
pdf_content: bytes,
|
||||
relative_path: str,
|
||||
storage_root: Path | None = None,
|
||||
) -> Path:
|
||||
"""Save PDF content to storage.
|
||||
|
||||
Args:
|
||||
pdf_content: Binary content of the PDF file.
|
||||
relative_path: Relative path (from compute_document_paths).
|
||||
storage_root: Optional custom storage root.
|
||||
|
||||
Returns:
|
||||
Absolute path to the saved file.
|
||||
"""
|
||||
if storage_root is None:
|
||||
storage_root = get_storage_root()
|
||||
|
||||
full_path = storage_root / relative_path
|
||||
full_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
full_path.write_bytes(pdf_content)
|
||||
return full_path
|
||||
|
||||
|
||||
def save_json(
|
||||
data: dict[str, Any],
|
||||
relative_path: str,
|
||||
storage_root: Path | None = None,
|
||||
) -> Path:
|
||||
"""Save JSON data to storage.
|
||||
|
||||
Args:
|
||||
data: Dictionary to serialize as JSON.
|
||||
relative_path: Relative path (from compute_document_paths).
|
||||
storage_root: Optional custom storage root.
|
||||
|
||||
Returns:
|
||||
Absolute path to the saved file.
|
||||
"""
|
||||
if storage_root is None:
|
||||
storage_root = get_storage_root()
|
||||
|
||||
full_path = storage_root / relative_path
|
||||
full_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
with open(full_path, "w", encoding="utf-8") as f:
|
||||
json.dump(data, f, ensure_ascii=False, indent=2, default=str)
|
||||
|
||||
return full_path
|
||||
|
||||
|
||||
def get_absolute_path(relative_path: str, storage_root: Path | None = None) -> Path:
|
||||
"""Get absolute path from relative storage path.
|
||||
|
||||
Args:
|
||||
relative_path: Relative path stored in database.
|
||||
storage_root: Optional custom storage root.
|
||||
|
||||
Returns:
|
||||
Absolute path to the file.
|
||||
"""
|
||||
if storage_root is None:
|
||||
storage_root = get_storage_root()
|
||||
return storage_root / relative_path
|
||||
|
||||
|
||||
def file_exists(relative_path: str, storage_root: Path | None = None) -> bool:
|
||||
"""Check if a file exists in storage.
|
||||
|
||||
Args:
|
||||
relative_path: Relative path stored in database.
|
||||
storage_root: Optional custom storage root.
|
||||
|
||||
Returns:
|
||||
True if file exists.
|
||||
"""
|
||||
return get_absolute_path(relative_path, storage_root).exists()
|
||||
|
||||
|
||||
def read_pdf(relative_path: str, storage_root: Path | None = None) -> bytes:
|
||||
"""Read PDF content from storage.
|
||||
|
||||
Args:
|
||||
relative_path: Relative path stored in database.
|
||||
storage_root: Optional custom storage root.
|
||||
|
||||
Returns:
|
||||
Binary content of the PDF.
|
||||
|
||||
Raises:
|
||||
FileNotFoundError: If file doesn't exist.
|
||||
"""
|
||||
full_path = get_absolute_path(relative_path, storage_root)
|
||||
return full_path.read_bytes()
|
||||
|
||||
|
||||
def read_json(relative_path: str, storage_root: Path | None = None) -> dict[str, Any]:
|
||||
"""Read JSON data from storage.
|
||||
|
||||
Args:
|
||||
relative_path: Relative path stored in database.
|
||||
storage_root: Optional custom storage root.
|
||||
|
||||
Returns:
|
||||
Parsed JSON data.
|
||||
|
||||
Raises:
|
||||
FileNotFoundError: If file doesn't exist.
|
||||
"""
|
||||
full_path = get_absolute_path(relative_path, storage_root)
|
||||
with open(full_path, "r", encoding="utf-8") as f:
|
||||
return json.load(f)
|
||||
|
||||
|
||||
def delete_document_files(
|
||||
pdf_path: str | None,
|
||||
json_path: str | None,
|
||||
storage_root: Path | None = None,
|
||||
) -> tuple[bool, bool]:
|
||||
"""Delete document files from storage.
|
||||
|
||||
Args:
|
||||
pdf_path: Relative path to PDF (can be None).
|
||||
json_path: Relative path to JSON (can be None).
|
||||
storage_root: Optional custom storage root.
|
||||
|
||||
Returns:
|
||||
Tuple of (pdf_deleted, json_deleted) booleans.
|
||||
"""
|
||||
if storage_root is None:
|
||||
storage_root = get_storage_root()
|
||||
|
||||
pdf_deleted = False
|
||||
json_deleted = False
|
||||
|
||||
if pdf_path:
|
||||
pdf_full = storage_root / pdf_path
|
||||
if pdf_full.exists():
|
||||
pdf_full.unlink()
|
||||
pdf_deleted = True
|
||||
|
||||
if json_path:
|
||||
json_full = storage_root / json_path
|
||||
if json_full.exists():
|
||||
json_full.unlink()
|
||||
json_deleted = True
|
||||
|
||||
return pdf_deleted, json_deleted
|
||||
Reference in New Issue
Block a user