feat(stock): support multi-page photo uploads for quote OCR processing
- Allow uploading multiple quote photos at once via multi-file input - Process photos sequentially with deskewing and OCR line clustering - Aggregate articles and metadata across all pages seamlessly - Attach all uploaded photos to the generated purchase order - Add unit tests for multi-image quotes and multiple document attachments
This commit is contained in:
parent
4010dfb138
commit
53d579009a
4 changed files with 401 additions and 128 deletions
|
|
@ -191,36 +191,42 @@ class PdfQuoteParser:
|
|||
|
||||
IMAGE_EXTENSIONS = ('.png', '.jpg', '.jpeg', '.webp', '.bmp', '.tiff', '.tif')
|
||||
|
||||
def __init__(self, file_or_path):
|
||||
self.file_or_path = file_or_path
|
||||
self._temp_path = None
|
||||
def __init__(self, files_or_paths):
|
||||
if isinstance(files_or_paths, (list, tuple)):
|
||||
self.files = list(files_or_paths)
|
||||
elif files_or_paths:
|
||||
self.files = [files_or_paths]
|
||||
else:
|
||||
self.files = []
|
||||
|
||||
self.file_or_path = self.files[0] if self.files else None
|
||||
self._temp_paths = []
|
||||
self.is_ocr = False
|
||||
self.is_image = False
|
||||
|
||||
def _is_image(self) -> bool:
|
||||
"""Détecte si le fichier fourni est une image ou une photo."""
|
||||
if isinstance(self.file_or_path, str):
|
||||
ext = os.path.splitext(self.file_or_path)[1].lower()
|
||||
def _is_file_image(self, target) -> bool:
|
||||
"""Vérifie si un fichier ou chemin cible est une image."""
|
||||
if isinstance(target, str):
|
||||
ext = os.path.splitext(target)[1].lower()
|
||||
if ext in self.IMAGE_EXTENSIONS:
|
||||
return True
|
||||
elif hasattr(self.file_or_path, 'name'):
|
||||
ext = os.path.splitext(self.file_or_path.name)[1].lower()
|
||||
elif hasattr(target, 'name'):
|
||||
ext = os.path.splitext(target.name)[1].lower()
|
||||
if ext in self.IMAGE_EXTENSIONS:
|
||||
return True
|
||||
|
||||
# Détection par signature d'octets si disponible
|
||||
if hasattr(self.file_or_path, 'read'):
|
||||
header = self.file_or_path.read(16)
|
||||
if hasattr(self.file_or_path, 'seek'):
|
||||
self.file_or_path.seek(0)
|
||||
if hasattr(target, 'read'):
|
||||
header = target.read(16)
|
||||
if hasattr(target, 'seek'):
|
||||
target.seek(0)
|
||||
if header.startswith(b'\x89PNG') or header.startswith(b'\xff\xd8\xff') or header.startswith(b'RIFF'):
|
||||
return True
|
||||
if header.startswith(b'%PDF'):
|
||||
return False
|
||||
|
||||
if isinstance(self.file_or_path, str) and os.path.exists(self.file_or_path):
|
||||
if isinstance(target, str) and os.path.exists(target):
|
||||
try:
|
||||
with open(self.file_or_path, 'rb') as f:
|
||||
with open(target, 'rb') as f:
|
||||
header = f.read(16)
|
||||
if header.startswith(b'\x89PNG') or header.startswith(b'\xff\xd8\xff') or header.startswith(b'RIFF'):
|
||||
return True
|
||||
|
|
@ -229,47 +235,92 @@ class PdfQuoteParser:
|
|||
|
||||
return False
|
||||
|
||||
def _is_image(self) -> bool:
|
||||
"""Détecte si les fichiers fournis sont des images ou des photos."""
|
||||
if not self.files:
|
||||
return False
|
||||
return all(self._is_file_image(f) for f in self.files)
|
||||
|
||||
def _get_document(self) -> Tuple[Any, str]:
|
||||
"""Ouvre le document PDF via pypdfium2."""
|
||||
"""Ouvre le ou les documents PDF via pypdfium2."""
|
||||
if pdfium is None:
|
||||
raise RuntimeError(
|
||||
"Le module 'pypdfium2' n'est pas installé sur le serveur. "
|
||||
"Veuillez exécuter 'pip install -r requirements/base.txt' pour activer l'analyse PDF locale."
|
||||
)
|
||||
if isinstance(self.file_or_path, str):
|
||||
return pdfium.PdfDocument(self.file_or_path), self.file_or_path
|
||||
elif hasattr(self.file_or_path, 'temporary_file_path'):
|
||||
path = self.file_or_path.temporary_file_path()
|
||||
return pdfium.PdfDocument(path), path
|
||||
elif hasattr(self.file_or_path, 'read'):
|
||||
content = self.file_or_path.read()
|
||||
if hasattr(self.file_or_path, 'seek'):
|
||||
self.file_or_path.seek(0)
|
||||
fd, tmp_path = tempfile.mkstemp(suffix='.pdf')
|
||||
with os.fdopen(fd, 'wb') as f:
|
||||
f.write(content)
|
||||
self._temp_path = tmp_path
|
||||
return pdfium.PdfDocument(tmp_path), tmp_path
|
||||
else:
|
||||
raise ValueError("Type de fichier non supporté pour l'analyse PDF.")
|
||||
if not self.files:
|
||||
raise ValueError("Aucun fichier fourni pour l'analyse PDF.")
|
||||
|
||||
if len(self.files) == 1:
|
||||
f = self.files[0]
|
||||
if isinstance(f, str):
|
||||
return pdfium.PdfDocument(f), f
|
||||
elif hasattr(f, 'temporary_file_path'):
|
||||
path = f.temporary_file_path()
|
||||
return pdfium.PdfDocument(path), path
|
||||
elif hasattr(f, 'read'):
|
||||
content = f.read()
|
||||
if hasattr(f, 'seek'):
|
||||
f.seek(0)
|
||||
fd, tmp_path = tempfile.mkstemp(suffix='.pdf')
|
||||
with os.fdopen(fd, 'wb') as out_f:
|
||||
out_f.write(content)
|
||||
self._temp_paths.append(tmp_path)
|
||||
return pdfium.PdfDocument(tmp_path), tmp_path
|
||||
else:
|
||||
raise ValueError("Type de fichier non supporté pour l'analyse PDF.")
|
||||
|
||||
# Plusieurs fichiers PDF : on fusionne leurs pages dans un document unique
|
||||
merged_doc = pdfium.PdfDocument.new()
|
||||
first_path = ""
|
||||
for f in self.files:
|
||||
sub_doc = None
|
||||
if isinstance(f, str):
|
||||
sub_doc = pdfium.PdfDocument(f)
|
||||
if not first_path:
|
||||
first_path = f
|
||||
elif hasattr(f, 'temporary_file_path'):
|
||||
path = f.temporary_file_path()
|
||||
sub_doc = pdfium.PdfDocument(path)
|
||||
if not first_path:
|
||||
first_path = path
|
||||
elif hasattr(f, 'read'):
|
||||
content = f.read()
|
||||
if hasattr(f, 'seek'):
|
||||
f.seek(0)
|
||||
fd, tmp_path = tempfile.mkstemp(suffix='.pdf')
|
||||
with os.fdopen(fd, 'wb') as out_f:
|
||||
out_f.write(content)
|
||||
self._temp_paths.append(tmp_path)
|
||||
sub_doc = pdfium.PdfDocument(tmp_path)
|
||||
if not first_path:
|
||||
first_path = tmp_path
|
||||
|
||||
if sub_doc:
|
||||
merged_doc.import_pages(sub_doc)
|
||||
sub_doc.close()
|
||||
|
||||
return merged_doc, first_path
|
||||
|
||||
def close(self):
|
||||
"""Nettoie les fichiers temporaires si besoin."""
|
||||
if self._temp_path and os.path.exists(self._temp_path):
|
||||
try:
|
||||
os.remove(self._temp_path)
|
||||
except OSError:
|
||||
pass
|
||||
for p in self._temp_paths:
|
||||
if p and os.path.exists(p):
|
||||
try:
|
||||
os.remove(p)
|
||||
except OSError:
|
||||
pass
|
||||
self._temp_paths = []
|
||||
|
||||
def parse(self) -> Dict[str, Any]:
|
||||
"""
|
||||
Extrait les métadonnées et la liste des articles d'un devis (PDF ou Image).
|
||||
Extrait les métadonnées et la liste des articles d'un devis (PDF ou Image(s)).
|
||||
Effectue le rapprochement automatique avec le stock existant (fournisseur et produits).
|
||||
"""
|
||||
if self._is_image():
|
||||
self.is_ocr = True
|
||||
self.is_image = True
|
||||
return self._parse_image()
|
||||
return self._parse_images()
|
||||
|
||||
doc = None
|
||||
try:
|
||||
|
|
@ -294,8 +345,8 @@ class PdfQuoteParser:
|
|||
pass
|
||||
self.close()
|
||||
|
||||
def _parse_image(self) -> Dict[str, Any]:
|
||||
"""Analyse directe d'une photo ou d'un scan d'offre via RapidOCR local."""
|
||||
def _parse_images(self) -> Dict[str, Any]:
|
||||
"""Analyse directe d'une ou plusieurs photos ou scans d'offre via RapidOCR local."""
|
||||
if Image is None:
|
||||
raise RuntimeError("Le module PIL/Pillow n'est pas disponible pour l'analyse d'images.")
|
||||
|
||||
|
|
@ -306,53 +357,62 @@ class PdfQuoteParser:
|
|||
logger.error("RapidOCR n'est pas disponible pour l'OCR local.")
|
||||
raise RuntimeError("Le module OCR local n'est pas installé dans l'environnement.")
|
||||
|
||||
# Charger l'image avec gestion de l'orientation EXIF des smartphones
|
||||
if isinstance(self.file_or_path, str):
|
||||
pil_img = Image.open(self.file_or_path)
|
||||
elif hasattr(self.file_or_path, 'temporary_file_path'):
|
||||
pil_img = Image.open(self.file_or_path.temporary_file_path())
|
||||
elif hasattr(self.file_or_path, 'read'):
|
||||
pil_img = Image.open(self.file_or_path)
|
||||
if hasattr(self.file_or_path, 'seek'):
|
||||
self.file_or_path.seek(0)
|
||||
else:
|
||||
raise ValueError("Type d'image non supporté.")
|
||||
ocr_pages_data = []
|
||||
full_text_lines = []
|
||||
|
||||
if ImageOps is not None:
|
||||
pil_img = ImageOps.exif_transpose(pil_img)
|
||||
pil_img = pil_img.convert('RGB')
|
||||
for p_idx, f_item in enumerate(self.files, start=1):
|
||||
if isinstance(f_item, str):
|
||||
pil_img = Image.open(f_item)
|
||||
elif hasattr(f_item, 'temporary_file_path'):
|
||||
pil_img = Image.open(f_item.temporary_file_path())
|
||||
elif hasattr(f_item, 'read'):
|
||||
pil_img = Image.open(f_item)
|
||||
if hasattr(f_item, 'seek'):
|
||||
f_item.seek(0)
|
||||
else:
|
||||
raise ValueError(f"Type d'image non supporté : {type(f_item)}")
|
||||
|
||||
# Redimensionnement maîtrisé si image très volumineuse (ex: photo smartphone 48MP)
|
||||
max_dim = max(pil_img.size)
|
||||
if max_dim > 2500:
|
||||
scale = 2500.0 / max_dim
|
||||
new_size = (int(pil_img.width * scale), int(pil_img.height * scale))
|
||||
pil_img = pil_img.resize(new_size, Image.Resampling.LANCZOS)
|
||||
# Charger l'image avec gestion de l'orientation EXIF des smartphones
|
||||
if ImageOps is not None:
|
||||
pil_img = ImageOps.exif_transpose(pil_img)
|
||||
pil_img = pil_img.convert('RGB')
|
||||
|
||||
pil_img, ocr_res = self._detect_and_deskew_image(pil_img, ocr)
|
||||
rects = []
|
||||
h = pil_img.height
|
||||
# Redimensionnement maîtrisé si image très volumineuse (ex: photo smartphone 48MP)
|
||||
max_dim = max(pil_img.size)
|
||||
if max_dim > 2500:
|
||||
scale = 2500.0 / max_dim
|
||||
new_size = (int(pil_img.width * scale), int(pil_img.height * scale))
|
||||
pil_img = pil_img.resize(new_size, Image.Resampling.LANCZOS)
|
||||
|
||||
if ocr_res:
|
||||
for item in ocr_res:
|
||||
box, txt, conf = item
|
||||
if not txt.strip():
|
||||
continue
|
||||
left = min(pt[0] for pt in box)
|
||||
right = max(pt[0] for pt in box)
|
||||
top = min(pt[1] for pt in box)
|
||||
bottom = max(pt[1] for pt in box)
|
||||
rects.append((left, h - bottom, right, h - top, txt.strip()))
|
||||
pil_img, ocr_res = self._detect_and_deskew_image(pil_img, ocr)
|
||||
rects = []
|
||||
h = pil_img.height
|
||||
|
||||
rects.sort(key=lambda x: -x[3])
|
||||
avg_h = float(np.mean([r[3] - r[1] for r in rects])) if rects else 15.0
|
||||
y_tol = max(6.0, avg_h * 0.40)
|
||||
if ocr_res:
|
||||
for item in ocr_res:
|
||||
box, txt, conf = item
|
||||
if not txt.strip():
|
||||
continue
|
||||
left = min(pt[0] for pt in box)
|
||||
right = max(pt[0] for pt in box)
|
||||
top = min(pt[1] for pt in box)
|
||||
bottom = max(pt[1] for pt in box)
|
||||
rects.append((left, h - bottom, right, h - top, txt.strip()))
|
||||
|
||||
page_lines = self._cluster_lines(rects, y_tol=y_tol)
|
||||
full_text = "\n".join(" ".join(r[4] for r in l) for l in page_lines)
|
||||
rects.sort(key=lambda x: -x[3])
|
||||
avg_h = float(np.mean([r[3] - r[1] for r in rects])) if rects else 15.0
|
||||
y_tol = max(6.0, avg_h * 0.40)
|
||||
|
||||
page_lines = self._cluster_lines(rects, y_tol=y_tol)
|
||||
ocr_pages_data.append(page_lines)
|
||||
for l in page_lines:
|
||||
full_text_lines.append(' '.join(item[4] for item in l))
|
||||
full_text_lines.append(f"--- Page {p_idx} ---")
|
||||
|
||||
full_text = "\n".join(full_text_lines)
|
||||
|
||||
metadata = self._extract_metadata(full_text)
|
||||
items = self._extract_items_from_clustered_lines([page_lines])
|
||||
items = self._extract_items_from_clustered_lines(ocr_pages_data)
|
||||
|
||||
matched_supplier = self._match_supplier(metadata)
|
||||
enriched_items = self._match_products(items)
|
||||
|
|
@ -360,12 +420,19 @@ class PdfQuoteParser:
|
|||
return {
|
||||
'is_ocr': True,
|
||||
'is_image': True,
|
||||
'is_multi_image': len(self.files) > 1,
|
||||
'image_count': len(self.files),
|
||||
'page_count': len(self.files),
|
||||
'metadata': metadata,
|
||||
'items': enriched_items,
|
||||
'matched_supplier': matched_supplier,
|
||||
'total_items': len(enriched_items),
|
||||
}
|
||||
|
||||
def _parse_image(self) -> Dict[str, Any]:
|
||||
"""Méthode de compatibilité pour analyse d'image unique."""
|
||||
return self._parse_images()
|
||||
|
||||
@staticmethod
|
||||
def _detect_and_deskew_image(pil_img: Any, ocr: Any) -> Tuple[Any, Any]:
|
||||
"""
|
||||
|
|
|
|||
|
|
@ -41,7 +41,7 @@
|
|||
</div>
|
||||
<h4 class="fw-bold">{% translate "Importer une offre ou un devis fournisseur" %}</h4>
|
||||
<p class="text-muted">
|
||||
{% translate "Déposez un devis au format PDF, un scan ou une photo prise avec un smartphone. Le système analyse automatiquement le document avec reconnaissance OCR locale, extrait les articles, compare avec le stock existant et prépare le bon de commande." %}
|
||||
{% translate "Déposez un devis au format PDF, un scan ou des photos prises avec un smartphone (sélectionnez toutes les pages en une fois). Le système combine et analyse automatiquement les pages avec reconnaissance OCR locale, extrait les articles et prépare le bon de commande." %}
|
||||
</p>
|
||||
</div>
|
||||
|
||||
|
|
@ -64,22 +64,22 @@
|
|||
</div>
|
||||
</div>
|
||||
|
||||
<!-- Zone de dépôt Document (PDF ou Image) -->
|
||||
<!-- Zone de dépôt Document (PDF ou Image(s)) -->
|
||||
<div class="mb-4">
|
||||
<label class="form-label fw-semibold">
|
||||
<i class="bi bi-file-earmark-text me-1"></i> {% translate "Fichier de l'offre (PDF, scan ou photo)" %} <span class="text-danger">*</span>
|
||||
<i class="bi bi-file-earmark-text me-1"></i> {% translate "Fichier(s) de l'offre (PDF, scan ou photos des pages)" %} <span class="text-danger">*</span>
|
||||
</label>
|
||||
<div class="border border-2 border-dashed rounded-3 p-4 text-center bg-light" id="drop-zone" style="cursor: pointer;">
|
||||
<div class="mb-2">
|
||||
<i class="bi bi-file-earmark-pdf text-danger fs-2 me-2"></i>
|
||||
<i class="bi bi-camera text-primary fs-2"></i>
|
||||
</div>
|
||||
<span class="fw-medium text-dark d-block mb-1">{% translate "Cliquez ou glissez-déposez le document ici" %}</span>
|
||||
<span class="text-muted small d-block mb-2">{% translate "Formats acceptés : PDF (numérique ou scanné), photos & scans (JPG, PNG, WEBP, TIFF)" %}</span>
|
||||
<span class="fw-medium text-dark d-block mb-1">{% translate "Cliquez ou glissez-déposez vos documents ou photos ici" %}</span>
|
||||
<span class="text-muted small d-block mb-2">{% translate "Formats acceptés : PDF, photos & scans (JPG, PNG, WEBP, TIFF) — Sélection multiple de photos autorisée pour plusieurs pages" %}</span>
|
||||
<span class="badge bg-secondary-subtle text-secondary border">
|
||||
<i class="bi bi-cpu me-1"></i>{% translate "Reconnaissance OCR 100% locale" %}
|
||||
</span>
|
||||
<input type="file" name="pdf_file" id="id_pdf_file" class="d-none" accept=".pdf,image/*,.png,.jpg,.jpeg,.webp,.bmp,.tiff,.tif" required>
|
||||
<input type="file" name="pdf_file" id="id_pdf_file" class="d-none" accept=".pdf,image/*,.png,.jpg,.jpeg,.webp,.bmp,.tiff,.tif" multiple required>
|
||||
<div id="selected-file-name" class="mt-3 fw-bold text-primary d-none">
|
||||
<i class="bi bi-file-earmark-check me-1"></i> <span class="file-text"></span>
|
||||
</div>
|
||||
|
|
@ -115,6 +115,7 @@
|
|||
<input type="hidden" name="warehouse_id" value="{{ selected_warehouse.id }}">
|
||||
<input type="hidden" name="temp_pdf_path" value="{{ temp_pdf_path }}">
|
||||
<input type="hidden" name="orig_pdf_name" value="{{ orig_pdf_name }}">
|
||||
<input type="hidden" name="temp_files_json" value="{{ temp_files_json }}">
|
||||
<input type="hidden" name="item_count" value="{{ parse_result.items|length }}">
|
||||
|
||||
<!-- Cartes d'en-tête (Fournisseur & Commande) -->
|
||||
|
|
@ -266,9 +267,17 @@
|
|||
<div class="small text-muted d-flex align-items-center flex-wrap gap-2">
|
||||
<div>
|
||||
<i class="bi bi-paperclip me-1 text-primary"></i>
|
||||
<span>{% translate "Pièce justificative attachée :" %} <strong>{{ orig_pdf_name }}</strong></span>
|
||||
{% if parse_result.is_multi_image %}
|
||||
<span>{% translate "Pièces justificatives attachées :" %} <strong>{{ parse_result.image_count }} {% translate "photos / pages" %}</strong></span>
|
||||
{% else %}
|
||||
<span>{% translate "Pièce justificative attachée :" %} <strong>{{ orig_pdf_name }}</strong></span>
|
||||
{% endif %}
|
||||
</div>
|
||||
{% if parse_result.is_image %}
|
||||
{% if parse_result.is_multi_image %}
|
||||
<span class="badge bg-primary-subtle text-primary border border-primary-subtle">
|
||||
<i class="bi bi-images me-1"></i>{% blocktranslate with count=parse_result.image_count %}{{ count }} photos combinées (OCR local){% endblocktranslate %}
|
||||
</span>
|
||||
{% elif parse_result.is_image %}
|
||||
<span class="badge bg-info-subtle text-info-emphasis border border-info-subtle">
|
||||
<i class="bi bi-camera me-1"></i>{% translate "Photo / Scan (OCR local)" %}
|
||||
</span>
|
||||
|
|
@ -282,6 +291,15 @@
|
|||
</span>
|
||||
{% endif %}
|
||||
</div>
|
||||
{% if saved_temp_files and saved_temp_files|length > 1 %}
|
||||
<div class="mt-2 d-flex flex-wrap gap-1">
|
||||
{% for f in saved_temp_files %}
|
||||
<span class="badge bg-light text-dark border">
|
||||
<i class="bi bi-file-earmark-image text-primary me-1"></i>{{ f.name }}
|
||||
</span>
|
||||
{% endfor %}
|
||||
</div>
|
||||
{% endif %}
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
|
|
@ -544,13 +562,39 @@ document.addEventListener('DOMContentLoaded', function() {
|
|||
|
||||
function updateFileDisplay() {
|
||||
if (fileInput.files.length > 0) {
|
||||
const file = fileInput.files[0];
|
||||
const isImg = file.type.startsWith('image/') || /\.(png|jpe?g|webp|bmp|tiff?)$/i.test(file.name);
|
||||
const count = fileInput.files.length;
|
||||
let totalBytes = 0;
|
||||
const fileNames = [];
|
||||
let hasImg = false;
|
||||
|
||||
for (let i = 0; i < count; i++) {
|
||||
const file = fileInput.files[i];
|
||||
totalBytes += file.size;
|
||||
fileNames.push(file.name);
|
||||
if (file.type.startsWith('image/') || /\.(png|jpe?g|webp|bmp|tiff?)$/i.test(file.name)) {
|
||||
hasImg = true;
|
||||
}
|
||||
}
|
||||
|
||||
const icon = selectedFileDiv.querySelector('i');
|
||||
if (icon) {
|
||||
icon.className = isImg ? 'bi bi-camera me-1 text-primary' : 'bi bi-file-earmark-pdf me-1 text-danger';
|
||||
if (count > 1) {
|
||||
icon.className = 'bi bi-images me-1 text-primary';
|
||||
} else {
|
||||
icon.className = hasImg ? 'bi bi-camera me-1 text-primary' : 'bi bi-file-earmark-pdf me-1 text-danger';
|
||||
}
|
||||
}
|
||||
|
||||
const totalSizeStr = (totalBytes / 1024 > 1024)
|
||||
? (totalBytes / (1024 * 1024)).toFixed(2) + ' MB'
|
||||
: (totalBytes / 1024).toFixed(1) + ' KB';
|
||||
|
||||
if (count === 1) {
|
||||
selectedFileDiv.querySelector('.file-text').textContent = fileNames[0] + ' (' + totalSizeStr + ')';
|
||||
} else {
|
||||
selectedFileDiv.querySelector('.file-text').textContent =
|
||||
count + ' {% translate "fichiers / photos sélectionnés" %} (' + totalSizeStr + ') : ' + fileNames.slice(0, 3).join(', ') + (count > 3 ? '...' : '');
|
||||
}
|
||||
selectedFileDiv.querySelector('.file-text').textContent = file.name + ' (' + (file.size / 1024).toFixed(1) + ' KB)';
|
||||
selectedFileDiv.classList.remove('d-none');
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2817,3 +2817,118 @@ class PurchaseOrderFromDocumentTests(TestCase):
|
|||
self.assertEqual(split[2]['quantity'], 1)
|
||||
self.assertEqual(split[2]['total_amount'], 71.84)
|
||||
|
||||
def test_parser_detects_multiple_images(self):
|
||||
from .pdf_parser import PdfQuoteParser
|
||||
parser = PdfQuoteParser(["page_1.png", "page_2.jpg", "page_3.jpeg"])
|
||||
self.assertTrue(parser._is_image())
|
||||
self.assertEqual(len(parser.files), 3)
|
||||
|
||||
# Mix with PDF is not purely image
|
||||
parser_mixed = PdfQuoteParser(["page_1.png", "doc.pdf"])
|
||||
self.assertFalse(parser_mixed._is_image())
|
||||
|
||||
def test_view_confirm_order_attaches_multiple_documents(self):
|
||||
import json
|
||||
import os
|
||||
import tempfile
|
||||
from django.urls import reverse
|
||||
from .models import PurchaseOrder, PurchaseOrderDocument
|
||||
|
||||
self.client.force_login(self.superuser)
|
||||
url = reverse("stock:purchase_order_from_pdf")
|
||||
|
||||
# Create two temporary files simulating 2 uploaded photo pages
|
||||
fd1, tmp_p1 = tempfile.mkstemp(prefix="test_photo1_", suffix=".jpg")
|
||||
with os.fdopen(fd1, "wb") as f:
|
||||
f.write(b"fake photo page 1")
|
||||
|
||||
fd2, tmp_p2 = tempfile.mkstemp(prefix="test_photo2_", suffix=".jpg")
|
||||
with os.fdopen(fd2, "wb") as f:
|
||||
f.write(b"fake photo page 2")
|
||||
|
||||
temp_files_json = json.dumps([
|
||||
{"path": tmp_p1, "name": "Offre_Page_1.jpg"},
|
||||
{"path": tmp_p2, "name": "Offre_Page_2.jpg"},
|
||||
])
|
||||
|
||||
data = {
|
||||
"action": "confirm_order",
|
||||
"warehouse_id": self.warehouse.pk,
|
||||
"supplier_choice": "existing",
|
||||
"supplier_id": self.supplier.pk,
|
||||
"temp_files_json": temp_files_json,
|
||||
"item_count": "1",
|
||||
"item_0_include": "on",
|
||||
"item_0_product_id": self.product.pk,
|
||||
"item_0_ref": "ZE 45525",
|
||||
"item_0_name": "Mètre ruban",
|
||||
"item_0_qty": "3",
|
||||
"item_0_price": "9.28",
|
||||
"item_0_vat": "21.00",
|
||||
"item_0_location": self.location.pk,
|
||||
}
|
||||
|
||||
response = self.client.post(url, data)
|
||||
self.assertEqual(response.status_code, 302)
|
||||
|
||||
po = PurchaseOrder.objects.filter(supplier=self.supplier).last()
|
||||
self.assertIsNotNone(po)
|
||||
|
||||
# Verify that both documents were attached to the purchase order
|
||||
docs = PurchaseOrderDocument.objects.filter(purchase_order=po).order_by("name")
|
||||
self.assertEqual(docs.count(), 2)
|
||||
doc_names = [d.name for d in docs]
|
||||
self.assertIn("Offre_Page_1.jpg", doc_names)
|
||||
self.assertIn("Offre_Page_2.jpg", doc_names)
|
||||
|
||||
# Verify temp files were deleted
|
||||
self.assertFalse(os.path.exists(tmp_p1))
|
||||
self.assertFalse(os.path.exists(tmp_p2))
|
||||
|
||||
def test_view_parse_multiple_uploaded_files(self):
|
||||
from unittest.mock import patch
|
||||
from django.core.files.uploadedfile import SimpleUploadedFile
|
||||
from django.urls import reverse
|
||||
|
||||
self.client.force_login(self.superuser)
|
||||
url = reverse("stock:purchase_order_from_pdf")
|
||||
|
||||
file1 = SimpleUploadedFile("page_1.png", b"\x89PNG\r\n\x1a\nfakeimage1", content_type="image/png")
|
||||
file2 = SimpleUploadedFile("page_2.png", b"\x89PNG\r\n\x1a\nfakeimage2", content_type="image/png")
|
||||
|
||||
mock_parse_result = {
|
||||
'is_ocr': True,
|
||||
'is_image': True,
|
||||
'is_multi_image': True,
|
||||
'image_count': 2,
|
||||
'total_items': 1,
|
||||
'metadata': {'supplier_name': 'Test Sup', 'supplier_vat': 'BE 0111.222.333', 'quote_number': 'OF123'},
|
||||
'matched_supplier': self.supplier,
|
||||
'items': [{
|
||||
'reference': 'ZE 45525',
|
||||
'name': 'Mètre ruban',
|
||||
'quantity': 2,
|
||||
'unit': 'piece',
|
||||
'raw_unit': 'Piece',
|
||||
'gross_unit_price': 9.28,
|
||||
'discount': '0%',
|
||||
'unit_price': 9.28,
|
||||
'total_amount': 18.56,
|
||||
'vat_rate': 21.0,
|
||||
'matched_product': self.product,
|
||||
}],
|
||||
}
|
||||
|
||||
with patch("stock.pdf_parser.PdfQuoteParser.parse", return_value=mock_parse_result):
|
||||
response = self.client.post(url, {
|
||||
"action": "parse_pdf",
|
||||
"warehouse": self.warehouse.pk,
|
||||
"pdf_file": [file1, file2],
|
||||
})
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertEqual(response.context["step"], "review")
|
||||
self.assertTrue(response.context["parse_result"]["is_multi_image"])
|
||||
self.assertEqual(response.context["parse_result"]["image_count"], 2)
|
||||
self.assertEqual(len(response.context["saved_temp_files"]), 2)
|
||||
|
||||
|
|
|
|||
|
|
@ -1602,6 +1602,8 @@ def purchase_order_from_pdf(request):
|
|||
pour approbation ultérieure par un superviseur) ou l'associe à un produit existant.
|
||||
"""
|
||||
import os
|
||||
import re
|
||||
import json
|
||||
import tempfile
|
||||
from common.models import UserConfig
|
||||
try:
|
||||
|
|
@ -1624,42 +1626,66 @@ def purchase_order_from_pdf(request):
|
|||
warehouse = None
|
||||
temp_pdf_path = ''
|
||||
orig_pdf_name = ''
|
||||
temp_files_json = '[]'
|
||||
saved_temp_files = []
|
||||
|
||||
if request.method == 'POST':
|
||||
action = request.POST.get('action', '')
|
||||
|
||||
if action == 'parse_pdf':
|
||||
doc_file = request.FILES.get('pdf_file') or request.FILES.get('document_file')
|
||||
doc_files = request.FILES.getlist('pdf_file') or request.FILES.getlist('document_file')
|
||||
warehouse_id = request.POST.get('warehouse')
|
||||
ALLOWED_EXTENSIONS = ('.pdf', '.png', '.jpg', '.jpeg', '.webp', '.bmp', '.tiff', '.tif')
|
||||
if not doc_file:
|
||||
messages.error(request, _("Veuillez sélectionner un document (PDF, scan ou photo)."))
|
||||
if not doc_files:
|
||||
messages.error(request, _("Veuillez sélectionner un document ou des photos (PDF, scan ou photo)."))
|
||||
else:
|
||||
file_ext = os.path.splitext(doc_file.name.lower())[1]
|
||||
if file_ext not in ALLOWED_EXTENSIONS:
|
||||
invalid_files = [f.name for f in doc_files if os.path.splitext(f.name.lower())[1] not in ALLOWED_EXTENSIONS]
|
||||
if invalid_files:
|
||||
messages.error(
|
||||
request,
|
||||
_("Format de fichier non supporté. Veuillez téléverser un PDF ou une image (JPG, PNG, WEBP, TIFF).")
|
||||
_("Format de fichier non supporté pour : %(files)s. Veuillez téléverser un PDF ou une image (JPG, PNG, WEBP, TIFF).")
|
||||
% {'files': ', '.join(invalid_files[:3])}
|
||||
)
|
||||
else:
|
||||
warehouse = Warehouse.objects.filter(id__in=accessible_warehouses, pk=warehouse_id).first()
|
||||
if not warehouse:
|
||||
warehouse = warehouses.first()
|
||||
|
||||
# Sauvegarder dans un fichier temporaire pour pouvoir l'attacher après validation
|
||||
fd, tmp_path = tempfile.mkstemp(prefix='loko_quote_', suffix=file_ext)
|
||||
with os.fdopen(fd, 'wb') as f:
|
||||
for chunk in doc_file.chunks():
|
||||
f.write(chunk)
|
||||
temp_pdf_path = tmp_path
|
||||
orig_pdf_name = doc_file.name
|
||||
# Trier naturellement les fichiers s'il y en a plusieurs (ex: page_1, page_2...)
|
||||
if len(doc_files) > 1:
|
||||
def _nat_key(f):
|
||||
name = getattr(f, 'name', '') or str(f)
|
||||
return [int(t) if t.isdigit() else t.lower() for t in re.split(r'(\d+)', name)]
|
||||
doc_files = sorted(doc_files, key=_nat_key)
|
||||
|
||||
saved_temp_files = []
|
||||
for f in doc_files:
|
||||
file_ext = os.path.splitext(f.name.lower())[1]
|
||||
fd, tmp_path = tempfile.mkstemp(prefix='loko_quote_', suffix=file_ext)
|
||||
with os.fdopen(fd, 'wb') as out_f:
|
||||
for chunk in f.chunks():
|
||||
out_f.write(chunk)
|
||||
saved_temp_files.append({'path': tmp_path, 'name': f.name})
|
||||
|
||||
temp_paths = [item['path'] for item in saved_temp_files]
|
||||
orig_names = [item['name'] for item in saved_temp_files]
|
||||
temp_pdf_path = temp_paths[0] if temp_paths else ''
|
||||
orig_pdf_name = orig_names[0] if orig_names else ''
|
||||
import json as _json
|
||||
temp_files_json = _json.dumps(saved_temp_files)
|
||||
|
||||
try:
|
||||
from .pdf_parser import PdfQuoteParser
|
||||
parser = PdfQuoteParser(temp_pdf_path)
|
||||
parser = PdfQuoteParser(temp_paths)
|
||||
parse_result = parser.parse()
|
||||
step = 'review'
|
||||
if parse_result.get('is_image'):
|
||||
if parse_result.get('is_multi_image'):
|
||||
messages.info(
|
||||
request,
|
||||
_("%(count)d photos de pages ont été combinées et analysées avec succès via l'OCR local.")
|
||||
% {'count': parse_result.get('image_count', len(temp_paths))}
|
||||
)
|
||||
elif parse_result.get('is_image'):
|
||||
messages.info(request, _("L'image/photo a été analysée avec succès via le moteur OCR local."))
|
||||
elif parse_result.get('is_ocr'):
|
||||
messages.info(request, _("Le document PDF scanné a été analysé avec succès via le moteur OCR local."))
|
||||
|
|
@ -1669,12 +1695,17 @@ def purchase_order_from_pdf(request):
|
|||
import logging
|
||||
logging.getLogger(__name__).exception("Erreur lors de l'analyse du document d'achat")
|
||||
messages.error(request, _("Erreur lors de l'analyse du document : %(err)s") % {'err': str(e)})
|
||||
if temp_pdf_path and os.path.exists(temp_pdf_path):
|
||||
try:
|
||||
os.remove(temp_pdf_path)
|
||||
except OSError:
|
||||
pass
|
||||
for p in temp_paths:
|
||||
if p and os.path.exists(p):
|
||||
try:
|
||||
os.remove(p)
|
||||
except OSError:
|
||||
pass
|
||||
temp_paths = []
|
||||
saved_temp_files = []
|
||||
temp_pdf_path = ''
|
||||
orig_pdf_name = ''
|
||||
temp_files_json = '[]'
|
||||
step = 'upload'
|
||||
|
||||
elif action == 'confirm_order':
|
||||
|
|
@ -1686,6 +1717,17 @@ def purchase_order_from_pdf(request):
|
|||
|
||||
temp_pdf_path = request.POST.get('temp_pdf_path', '')
|
||||
orig_pdf_name = request.POST.get('orig_pdf_name', '')
|
||||
temp_files_json = request.POST.get('temp_files_json', '')
|
||||
|
||||
files_to_attach = []
|
||||
if temp_files_json:
|
||||
try:
|
||||
import json as _json
|
||||
files_to_attach = _json.loads(temp_files_json)
|
||||
except Exception:
|
||||
files_to_attach = []
|
||||
if not files_to_attach and temp_pdf_path:
|
||||
files_to_attach = [{'path': temp_pdf_path, 'name': orig_pdf_name}]
|
||||
|
||||
# Résolution ou création du Fournisseur
|
||||
supplier_choice = request.POST.get('supplier_choice', 'existing')
|
||||
|
|
@ -1790,21 +1832,24 @@ def purchase_order_from_pdf(request):
|
|||
)
|
||||
return redirect('stock:purchase_order_from_pdf')
|
||||
|
||||
# Attacher le fichier PDF original comme document justificatif
|
||||
if temp_pdf_path and os.path.exists(temp_pdf_path):
|
||||
try:
|
||||
from django.core.files.base import ContentFile
|
||||
with open(temp_pdf_path, 'rb') as f:
|
||||
doc_title = orig_pdf_name or f"Offre_{order.code}.pdf"
|
||||
po_doc = PurchaseOrderDocument(
|
||||
purchase_order=order,
|
||||
name=doc_title,
|
||||
)
|
||||
po_doc.file.save(doc_title, ContentFile(f.read()), save=True)
|
||||
os.remove(temp_pdf_path)
|
||||
except Exception:
|
||||
import logging
|
||||
logging.getLogger(__name__).exception("Erreur lors de l'attachement du PDF au bon de commande")
|
||||
# Attacher le ou les fichiers originaux comme pièces justificatives (PurchaseOrderDocument)
|
||||
for idx, file_info in enumerate(files_to_attach, start=1):
|
||||
f_path = file_info.get('path', '')
|
||||
f_name = file_info.get('name', '')
|
||||
if f_path and os.path.exists(f_path):
|
||||
try:
|
||||
from django.core.files.base import ContentFile
|
||||
with open(f_path, 'rb') as f:
|
||||
doc_title = f_name or f"Offre_{order.code}_p{idx}.pdf"
|
||||
po_doc = PurchaseOrderDocument(
|
||||
purchase_order=order,
|
||||
name=doc_title,
|
||||
)
|
||||
po_doc.file.save(doc_title, ContentFile(f.read()), save=True)
|
||||
os.remove(f_path)
|
||||
except Exception:
|
||||
import logging
|
||||
logging.getLogger(__name__).exception("Erreur lors de l'attachement du document au bon de commande")
|
||||
|
||||
messages.success(
|
||||
request,
|
||||
|
|
@ -1846,6 +1891,8 @@ def purchase_order_from_pdf(request):
|
|||
'parse_result': parse_result,
|
||||
'temp_pdf_path': temp_pdf_path,
|
||||
'orig_pdf_name': orig_pdf_name,
|
||||
'temp_files_json': temp_files_json,
|
||||
'saved_temp_files': saved_temp_files,
|
||||
'cancel_url': reverse('stock:stock_purchases'),
|
||||
})
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue