476 lines
19 KiB
Python
476 lines
19 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Simple Prompt Tester - Test current prompt without optimization
|
|
"""
|
|
import click
|
|
import json
|
|
import logging
|
|
import json
|
|
import PyPDF2
|
|
import os
|
|
|
|
from dotenv import load_dotenv
|
|
from pathlib import Path
|
|
from typing import List, Dict, Tuple
|
|
from datetime import datetime
|
|
from difflib import SequenceMatcher
|
|
from azure.ai.formrecognizer import DocumentAnalysisClient
|
|
from azure.core.credentials import AzureKeyCredential
|
|
from openai import AzureOpenAI
|
|
|
|
load_dotenv()
|
|
|
|
# Configuration Azure Document Intelligence
|
|
AZURE_DOC_ENDPOINT = os.getenv("AZURE_DOC_ENDPOINT")
|
|
AZURE_DOC_KEY = os.getenv("AZURE_DOC_KEY")
|
|
|
|
# Configuration Azure OpenAI
|
|
AZURE_OPENAI_ENDPOINT = os.getenv("AZURE_OPENAI_ENDPOINT")
|
|
AZURE_OPENAI_KEY = os.getenv("AZURE_OPENAI_KEY")
|
|
AZURE_DEPLOYMENT = os.getenv("AZURE_DEPLOYMENT", "gpt-4.1")
|
|
API_VERSION = os.getenv("API_VERSION", "2024-12-01-preview")
|
|
|
|
if not all([AZURE_DOC_ENDPOINT, AZURE_DOC_KEY, AZURE_OPENAI_ENDPOINT, AZURE_OPENAI_KEY]):
|
|
raise ValueError(
|
|
"❌ Missing Azure credentials! Please check your .env file.\n"
|
|
"Required variables: AZURE_DOC_ENDPOINT, AZURE_DOC_KEY, AZURE_OPENAI_ENDPOINT, AZURE_OPENAI_KEY"
|
|
)
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s - %(levelname)s - %(message)s'
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
logger.info("✅ Environment variables loaded successfully")
|
|
logger.info(f"📍 Document Intelligence endpoint: {AZURE_DOC_ENDPOINT}")
|
|
logger.info(f"🤖 OpenAI deployment: {AZURE_DEPLOYMENT}")
|
|
|
|
class PromptTester:
|
|
"""Test a single prompt on PDFs"""
|
|
def __init__(self):
|
|
self.doc_client = DocumentAnalysisClient(
|
|
endpoint=AZURE_DOC_ENDPOINT,
|
|
credential=AzureKeyCredential(AZURE_DOC_KEY)
|
|
)
|
|
self.openai_client = AzureOpenAI(
|
|
azure_endpoint=AZURE_OPENAI_ENDPOINT,
|
|
api_key=AZURE_OPENAI_KEY,
|
|
api_version=API_VERSION
|
|
)
|
|
self.ocr_cache = {}
|
|
|
|
def get_current_prompt(self) -> Tuple[str, str]:
|
|
"""Get the current working prompt"""
|
|
system_prompt = """Tu es un expert géotechnicien spécialisé dans l'analyse de logs de forage.
|
|
Ta tâche est d'extraire UNIQUEMENT les couches géologiques avec leurs profondeurs et descriptions COMPLÈTES.
|
|
|
|
RÈGLES CRITIQUES:
|
|
1. Chaque couche DOIT avoir une profondeur de début et de fin en mètres
|
|
2. La description doit contenir TOUT le texte associé à la couche (même sur plusieurs lignes)
|
|
3. Notation décimale: 0,3 ou 0.3 = 0.3m (PAS 3m!)
|
|
4. Profondeurs typiques: 0-0.3m, 0.3-1m, 1-4m, 4-10m, etc.
|
|
5. NE PAS séparer ou classifier le matériau - tout dans la description
|
|
6. IGNORER les mots isolés qui ne sont pas dans la colonne description
|
|
|
|
RÈGLE DE CONTINUITÉ DES COUCHES:
|
|
⚠️ Une couche géologique est CONTINUE entre deux profondeurs
|
|
⚠️ Si tu vois une profondeur isolée suivie de description, cherche OBLIGATOIREMENT la profondeur de fin
|
|
⚠️ Les profondeurs de fin peuvent être:
|
|
- Sur la même ligne après la description
|
|
- Sur une ligne suivante
|
|
- Au début de la couche suivante (la fin d'une couche = début de la suivante)
|
|
⚠️ NE JAMAIS fragmenter une description continue en plusieurs couches
|
|
|
|
GESTION DES PAGES:
|
|
⚠️ Les couches peuvent continuer d'une page à l'autre
|
|
⚠️ Sur une nouvelle page, si la première profondeur n'est pas 0.0m, c'est la continuation de la page précédente
|
|
⚠️ Ne pas dupliquer les couches identiques sur plusieurs pages
|
|
|
|
SORTIE: JSON valide uniquement"""
|
|
|
|
user_prompt_template = """Extrait toutes les couches géologiques de cette page de log de forage.
|
|
|
|
CE QUI EST UNE COUCHE GÉOLOGIQUE:
|
|
✓ Descriptions de matériaux: sol, roche, sable, argile, limon, remblai, terre végétale, etc.
|
|
✓ A une plage de profondeur ET des propriétés de matériau
|
|
✓ Décrit les caractéristiques physiques: couleur, consistance, humidité, altération, granulométrie
|
|
|
|
CE QUI N'EST PAS UNE COUCHE (IGNORER):
|
|
✗ Descriptions d'images ou légendes
|
|
✗ Métadonnées du projet (coordonnées, dates, noms d'entreprise)
|
|
✗ Descriptions d'équipement ou de méthodologie
|
|
✗ Données administratives ou commentaires sur la procédure
|
|
✗ Mots isolés dans d'autres colonnes (classifications, formations géologiques, etc.)
|
|
|
|
EXCLUSIONS ABSOLUES - NE JAMAIS extraire:
|
|
✗ Annotations techniques isolées: CZ, JT, JTx5, etc.
|
|
✗ Mesures techniques sans description de matériau
|
|
✗ Échantillons ("Undisturbed sample", "Disturbed sample", "sample", "échantillon")
|
|
✗ Lignes contenant uniquement des angles, pressions ou mesures
|
|
✗ Références à des essais ou tests sans description de sol
|
|
|
|
ALGORITHME DE DÉTECTION DES PROFONDEURS:
|
|
1. Pour chaque nombre qui ressemble à une profondeur (X.X m ou X,X m):
|
|
- Si suivi de "-" ou "à" ou "bis" → profondeur de début, chercher la fin
|
|
- Si précédé de "-" ou "à" ou "bis" → profondeur de fin
|
|
- Si isolé → vérifier le contexte:
|
|
* Début de description → profondeur de début
|
|
* Fin de description → profondeur de fin
|
|
* Nouvelle ligne → début de nouvelle couche
|
|
2. TOUJOURS vérifier la cohérence: fin couche N = début couche N+1
|
|
3. Si une couche n'a pas de profondeur de fin explicite, elle va jusqu'au début de la couche suivante
|
|
|
|
MOTIFS À RECHERCHER:
|
|
- "0.00-0.30m: Terre végétale, brun foncé..."
|
|
- "0,3 à 1,0m Sable fin beige..."
|
|
- "de 1m à 4m: Argile plastique grise..."
|
|
- "1.00 - 4.00 ARGILE limoneuse..."
|
|
- "0.3 m Description... 4.0 m" (profondeurs séparées)
|
|
- Tableaux avec colonnes profondeur/description
|
|
|
|
IDENTIFICATION DES COLONNES DE TABLEAU:
|
|
⚠️ Dans un tableau, identifier VISUELLEMENT les colonnes avant d'extraire
|
|
⚠️ Colonnes à IGNORER complètement:
|
|
- Formation géologique (Quaternaire, Tertiaire, Quaternaire, etc.)
|
|
- Stratigraphie
|
|
- Classification (Colluvions, Alluvions, etc.)
|
|
- Symboles ou codes isolés
|
|
- Âge géologique
|
|
⚠️ Colonnes à UTILISER:
|
|
- Description du matériau
|
|
- Propriétés physiques (couleur, texture, consistance)
|
|
⚠️ Si doute sur une colonne, vérifier si elle contient des propriétés physiques détaillées
|
|
|
|
RÈGLE CRITIQUE DE SÉPARATION:
|
|
⚠️ Les profondeurs (0-0.3m, 0.30m, etc.) ne font JAMAIS partie de la description
|
|
⚠️ Si une ligne contient "0,30 m - Mutterboden, braun", alors:
|
|
- depth_start: 0.0, depth_end: 0.3
|
|
- description: "Mutterboden, braun" (SANS les profondeurs)
|
|
⚠️ Retirer TOUS les préfixes de profondeur du champ description
|
|
⚠️ NE PAS inclure les profondeurs répétées à la fin de la description
|
|
|
|
IMPORTANT:
|
|
- Capture TOUT le texte de description, même s'il est long ou sur plusieurs lignes
|
|
- Ne pas résumer ou reformuler - garder le texte original
|
|
- Inclure TOUTES les propriétés mentionnées (couleur, texture, humidité, inclusions, etc.)
|
|
- EXCLURE ABSOLUMENT les profondeurs du champ description
|
|
- Si le texte commence par "0,30 m - Mutterboden", la description est SEULEMENT "Mutterboden"
|
|
- Séparer clairement: profondeurs dans depth_start/depth_end, matériau et propriétés dans description
|
|
- NE PAS mélanger les colonnes d'un tableau - prendre uniquement la colonne description principale
|
|
|
|
VALIDATION OBLIGATOIRE avant de retourner le JSON:
|
|
□ Toutes les couches ont une profondeur de début ET de fin
|
|
□ Les profondeurs sont continues (pas de trous entre les couches)
|
|
□ Les profondeurs sont cohérentes (début < fin)
|
|
□ Les descriptions ne contiennent PAS les profondeurs
|
|
□ Les descriptions ne contiennent PAS de noms de colonnes de classification
|
|
□ Aucune annotation technique isolée n'est présente
|
|
□ Pas de duplication de couches identiques
|
|
|
|
Retourne ce JSON:
|
|
{{
|
|
"layers": [
|
|
{{
|
|
"depth_start": 0.0,
|
|
"depth_end": 0.3,
|
|
"description": "Texte de description SANS les profondeurs - uniquement matériau et propriétés"
|
|
}}
|
|
]
|
|
}}
|
|
|
|
EXEMPLE CONCRET:
|
|
Texte original: "0,00 - 0,30 m Mutterboden, dunkelbraun, humos"
|
|
Résultat attendu:
|
|
{{
|
|
"depth_start": 0.0,
|
|
"depth_end": 0.3,
|
|
"description": "Mutterboden, dunkelbraun, humos"
|
|
}}
|
|
PAS: "description": "0,00 - 0,30 m Mutterboden, dunkelbraun, humos" ❌
|
|
|
|
TEXTE DE LA PAGE {page_num}:
|
|
{text}"""
|
|
|
|
return system_prompt, user_prompt_template
|
|
|
|
def extract_text_from_pdf(self, pdf_path: Path) -> List[str]:
|
|
"""Extract text from PDF with chunking"""
|
|
logger.info(f"🔍 Extracting text from {pdf_path.name}")
|
|
|
|
total_pages = self.get_page_count(pdf_path)
|
|
logger.info(f"📄 PDF has {total_pages} pages")
|
|
|
|
if total_pages == 0:
|
|
logger.error("Could not determine page count")
|
|
return []
|
|
|
|
# Extraire par chunks de 2 pages
|
|
pages_text = []
|
|
chunk_size = 2
|
|
for start_page in range(1, total_pages + 1, chunk_size):
|
|
end_page = min(start_page + chunk_size - 1, total_pages)
|
|
with open(pdf_path, "rb") as f:
|
|
poller = self.doc_client.begin_analyze_document(
|
|
model_id="prebuilt-document",
|
|
document=f,
|
|
pages=f"{start_page}-{end_page}"
|
|
)
|
|
result = poller.result()
|
|
for page in result.pages:
|
|
page_lines = [line.content for line in (page.lines or [])]
|
|
pages_text.append('\n'.join(page_lines))
|
|
|
|
return pages_text
|
|
|
|
def get_page_count(self, pdf_path: Path) -> int:
|
|
"""Get the number of pages in a PDF"""
|
|
try:
|
|
with open(pdf_path, "rb") as f:
|
|
pdf_reader = PyPDF2.PdfReader(f)
|
|
return len(pdf_reader.pages)
|
|
except:
|
|
pass
|
|
return 0
|
|
|
|
def _extract_table_text(self, table) -> str:
|
|
"""Extract text from table"""
|
|
if not hasattr(table, 'cells') or not table.cells:
|
|
return ""
|
|
|
|
max_row = max(cell.row_index for cell in table.cells)
|
|
max_col = max(cell.column_index for cell in table.cells)
|
|
table_array = [["" for _ in range(max_col + 1)] for _ in range(max_row + 1)]
|
|
|
|
for cell in table.cells:
|
|
if hasattr(cell, 'content'):
|
|
table_array[cell.row_index][cell.column_index] = cell.content
|
|
|
|
lines = []
|
|
for row in table_array:
|
|
row_text = " | ".join(str(cell).strip() for cell in row)
|
|
if row_text.strip():
|
|
lines.append(row_text)
|
|
return "\n".join(lines)
|
|
|
|
def test_prompt_on_pdf(self, system_prompt: str, user_prompt_template: str, pdf_path: Path) -> List[Dict]:
|
|
"""Test prompt on a single PDF"""
|
|
pages_text = self.extract_text_from_pdf(pdf_path)
|
|
all_layers = []
|
|
|
|
for page_idx, page_text in enumerate(pages_text):
|
|
if not page_text.strip():
|
|
continue
|
|
|
|
user_prompt = user_prompt_template.format(
|
|
page_num=page_idx + 1,
|
|
text=page_text[:4000]
|
|
)
|
|
|
|
try:
|
|
response = self.openai_client.chat.completions.create(
|
|
model=AZURE_DEPLOYMENT,
|
|
messages=[
|
|
{"role": "system", "content": system_prompt},
|
|
{"role": "user", "content": user_prompt}
|
|
],
|
|
temperature=0.1,
|
|
max_tokens=3000,
|
|
response_format={"type": "json_object"}
|
|
)
|
|
|
|
result = json.loads(response.choices[0].message.content)
|
|
|
|
for layer in result.get("layers", []):
|
|
try:
|
|
depth_start = float(str(layer.get("depth_start", 0)).replace(",", "."))
|
|
depth_end = float(str(layer.get("depth_end", 0)).replace(",", "."))
|
|
description = layer.get("description", "").strip()
|
|
|
|
if (depth_start < depth_end and depth_end <= 100 and
|
|
depth_start >= 0 and description):
|
|
all_layers.append({
|
|
"page": page_idx + 1,
|
|
"depth_start": depth_start,
|
|
"depth_end": depth_end,
|
|
"description": description
|
|
})
|
|
except:
|
|
continue
|
|
|
|
except Exception as e:
|
|
logger.error(f"GPT extraction failed for page {page_idx + 1}: {e}")
|
|
|
|
return sorted(all_layers, key=lambda x: (x['page'], x['depth_start']))
|
|
|
|
def compare_results(self, extracted: List[Dict], expected: List[Dict]) -> Dict:
|
|
"""Compare extracted layers with expected results"""
|
|
exact_matches = 0
|
|
depth_matches = 0
|
|
|
|
for exp in expected:
|
|
for ext in extracted:
|
|
if (ext['page'] == exp['page'] and
|
|
abs(ext['depth_start'] - exp['depth_start']) < 0.01 and
|
|
abs(ext['depth_end'] - exp['depth_end']) < 0.01):
|
|
depth_matches += 1
|
|
|
|
desc_sim = SequenceMatcher(None,
|
|
ext['description'].lower(),
|
|
exp['description'].lower()
|
|
).ratio()
|
|
|
|
if desc_sim == 1:
|
|
exact_matches += 1
|
|
break
|
|
|
|
if len(expected) > 0:
|
|
similarity = len([e for e in extracted if any(
|
|
abs(e['depth_start'] - x['depth_start']) < 0.5 and
|
|
abs(e['depth_end'] - x['depth_end']) < 0.5 and
|
|
e['page'] == x['page'] for x in expected
|
|
)]) / len(expected)
|
|
else:
|
|
similarity = 1.0 if len(extracted) == 0 else 0.0
|
|
|
|
return {
|
|
'similarity': similarity,
|
|
'exact_matches': exact_matches,
|
|
'depth_matches': depth_matches,
|
|
'extracted_count': len(extracted),
|
|
'expected_count': len(expected)
|
|
}
|
|
|
|
def test_all_pdfs(self, examples_dir: Path, expected_dir: Path):
|
|
"""Test prompt on all PDFs"""
|
|
test_pairs = []
|
|
for json_file in expected_dir.glob("*_layers.json"):
|
|
pdf_name = json_file.stem.replace("_layers", "") + ".pdf"
|
|
pdf_path = examples_dir / pdf_name
|
|
if pdf_path.exists():
|
|
test_pairs.append((pdf_path, json_file))
|
|
|
|
if not test_pairs:
|
|
click.echo("❌ No valid PDF/JSON pairs found!")
|
|
return
|
|
|
|
click.echo(f"\n📚 Found {len(test_pairs)} test PDFs")
|
|
|
|
system_prompt, user_prompt_template = self.get_current_prompt()
|
|
|
|
results = {}
|
|
total_similarity = 0
|
|
|
|
with click.progressbar(test_pairs, label='Testing PDFs') as pairs:
|
|
for pdf_path, json_path in pairs:
|
|
try:
|
|
with open(json_path, 'r', encoding='utf-8') as f:
|
|
data = json.load(f)
|
|
|
|
if isinstance(data, dict) and 'layers' in data:
|
|
expected_layers = data['layers']
|
|
elif isinstance(data, list):
|
|
expected_layers = data
|
|
else:
|
|
logger.warning(f"Unknown format in {json_path}")
|
|
continue
|
|
|
|
extracted_layers = self.test_prompt_on_pdf(
|
|
system_prompt, user_prompt_template, pdf_path
|
|
)
|
|
|
|
comparison = self.compare_results(extracted_layers, expected_layers)
|
|
results[pdf_path.name] = comparison
|
|
total_similarity += comparison['similarity']
|
|
|
|
single_result = {
|
|
"metadata": {
|
|
"source_file": str(pdf_path.resolve()),
|
|
"expected_file": str(json_path.resolve()),
|
|
"extraction_time": datetime.now().isoformat(),
|
|
"extracted_count": comparison['extracted_count'],
|
|
"expected_count": comparison['expected_count']
|
|
},
|
|
"layers": extracted_layers
|
|
}
|
|
|
|
output_dir = Path("output")
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
output_file_path = output_dir / f"{pdf_path.stem}_test_result.json"
|
|
|
|
with open(output_file_path, 'w', encoding='utf-8') as out_f:
|
|
json.dump(single_result, out_f, indent=2, ensure_ascii=False)
|
|
|
|
click.echo(f"💾 Résultat sauvegardé : {output_file_path}")
|
|
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error with {pdf_path.name}: {e}")
|
|
results[pdf_path.name] = {
|
|
'similarity': 0,
|
|
'error': str(e)
|
|
}
|
|
|
|
click.echo("\n" + "="*60)
|
|
click.echo("📊 TEST RESULTS")
|
|
click.echo("="*60)
|
|
|
|
for pdf_name, result in results.items():
|
|
if 'error' in result:
|
|
click.echo(f"\n❌ {pdf_name}: ERROR - {result['error']}")
|
|
else:
|
|
click.echo(f"\n📄 {pdf_name}:")
|
|
click.echo(f" Similarity: {result['similarity']:.1%}")
|
|
click.echo(f" Extracted: {result['extracted_count']} layers")
|
|
click.echo(f" Expected: {result['expected_count']} layers")
|
|
click.echo(f" Exact matches: {result['exact_matches']}")
|
|
click.echo(f" Depth matches: {result['depth_matches']}")
|
|
|
|
avg_similarity = total_similarity / len(test_pairs) if test_pairs else 0
|
|
click.echo("\n" + "="*60)
|
|
click.echo(f"📈 OVERALL: {avg_similarity:.1%} average similarity")
|
|
click.echo("="*60)
|
|
|
|
output_file = Path("output/test_results.json")
|
|
with open(output_file, 'w', encoding='utf-8') as f:
|
|
json.dump({
|
|
'timestamp': datetime.now().isoformat(),
|
|
'average_similarity': avg_similarity,
|
|
'results': results
|
|
}, f, indent=2)
|
|
|
|
click.echo(f"\n💾 Detailed results saved to: {output_file}")
|
|
|
|
@click.command()
|
|
@click.option('--examples-dir', '-e', default='examples',help='Directory containing test PDFs')
|
|
@click.option('--expected-dir', '-x', default='output_man', help='Directory containing expected results')
|
|
@click.option('--pdf', '-p', help='Test a single PDF only')
|
|
def main(examples_dir, expected_dir, pdf):
|
|
"""Test current prompt on PDFs without optimization"""
|
|
|
|
click.echo("🧪 Simple Prompt Tester")
|
|
click.echo("=" * 30)
|
|
|
|
tester = PromptTester()
|
|
|
|
if pdf:
|
|
pdf_path = Path(pdf)
|
|
if not pdf_path.exists():
|
|
raise click.ClickException(f"PDF not found: {pdf}")
|
|
|
|
click.echo(f"\n📄 Testing single PDF: {pdf_path.name}")
|
|
|
|
system_prompt, user_prompt_template = tester.get_current_prompt()
|
|
layers = tester.test_prompt_on_pdf(system_prompt, user_prompt_template, pdf_path)
|
|
|
|
click.echo(f"\n✅ Extracted {len(layers)} layers:")
|
|
for i, layer in enumerate(layers, 1):
|
|
click.echo(f"\n{i}. Depth: {layer['depth_start']:.2f} - {layer['depth_end']:.2f}m")
|
|
click.echo(f" Page: {layer['page']}")
|
|
click.echo(f" Description: {layer['description']}")
|
|
else:
|
|
tester.test_all_pdfs(Path(examples_dir), Path(expected_dir))
|
|
|
|
if __name__ == '__main__':
|
|
main()
|