Téléverser les fichiers vers "/"
This commit is contained in:
parent
09ffa4333b
commit
232d585021
BIN
69131.xlsx
Normal file
BIN
69131.xlsx
Normal file
Binary file not shown.
106
README.md
106
README.md
|
|
@ -1,3 +1,105 @@
|
||||||
# Structuredeprojet
|
# 🔧 Pole Engineering Project Automation
|
||||||
|
|
||||||
Créer le dossier du projets
|
Automatisation complète du traitement des projets de poteaux utilitaires.
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## ⚡ Installation rapide
|
||||||
|
|
||||||
|
```bash
|
||||||
|
pip install pillow openpyxl reportlab google-generativeai
|
||||||
|
```
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## 🚀 Utilisation
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# Utilisation minimale
|
||||||
|
python pole_automation.py E65-BK7-26-132679
|
||||||
|
|
||||||
|
# Avec chemins personnalisés
|
||||||
|
python pole_automation.py E65-BK7-26-132679 \
|
||||||
|
--base-dir /Users/moi/Projets \
|
||||||
|
--depart-dir /Users/moi/départ \
|
||||||
|
--gemini-key AIza...VotreCléAPI
|
||||||
|
```
|
||||||
|
|
||||||
|
### Variable d'environnement (recommandé pour la clé API)
|
||||||
|
```bash
|
||||||
|
export GEMINI_API_KEY="AIza...VotreCléAPI"
|
||||||
|
python pole_automation.py E65-BK7-26-132679
|
||||||
|
```
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## 📁 Structure des dossiers
|
||||||
|
|
||||||
|
### Dossier **départ** (source)
|
||||||
|
```
|
||||||
|
départ/
|
||||||
|
1d1-PBO345 rue saint-eugene/
|
||||||
|
photo_poteau.jpg
|
||||||
|
photo_ancrage.jpg
|
||||||
|
1d1-PBO345.xlsx
|
||||||
|
2sd2-PBO678 boul. des erables/
|
||||||
|
...
|
||||||
|
```
|
||||||
|
|
||||||
|
### Convention de nommage des sous-dossiers
|
||||||
|
| Partie | Exemple | Signification |
|
||||||
|
|--------------|---------|----------------------------------|
|
||||||
|
| `1` | `1` | Numéro de séquence |
|
||||||
|
| `d` ou `sd` | `d` | Avec (`d`) ou Sans dérogation (`sd`) |
|
||||||
|
| `1` | `1` | Nombre d'ancrages |
|
||||||
|
| `-PBO345` | `-PBO345` | Référence du poteau |
|
||||||
|
| `rue saint…` | _(optionnel)_ | Adresse |
|
||||||
|
|
||||||
|
### Projet créé automatiquement
|
||||||
|
```
|
||||||
|
<base-dir>/E65-BK7-26-132679/
|
||||||
|
compilation/ ← Tous les PDFs finaux ici
|
||||||
|
ingénierie/
|
||||||
|
RSA-UDS/ ← Dossiers renommés depuis départ
|
||||||
|
plan/
|
||||||
|
```
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## 🔄 Logique de renommage
|
||||||
|
|
||||||
|
| Type | Nouveau nom | Prochain saut |
|
||||||
|
|---------|---------------------------|---------------|
|
||||||
|
| `sd` | `10-PBO345` | +1 |
|
||||||
|
| `d` | `11-PBO345 rue saint-x` | +2 (CDC) |
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## 🤖 Intégration Gemini AI
|
||||||
|
|
||||||
|
Le script envoie les images JPG de chaque dossier à Gemini qui identifie :
|
||||||
|
- **Poteau** → converti en PDF avec le numéro séquentiel (ex: `12-PBO345.pdf`)
|
||||||
|
- **Ancrage** → converti en PDF avec le numéro suivant (ex: `13-PBO345.pdf`)
|
||||||
|
|
||||||
|
**Sans clé Gemini :** un algorithme de fallback classe les images par nom de fichier.
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## 📄 Fichiers PDF générés (dans `compilation/`)
|
||||||
|
|
||||||
|
Pour chaque poteau, dans l'ordre :
|
||||||
|
1. `{seq}-{ref}.pdf` ← Images du **poteau**
|
||||||
|
2. `{seq}-{ref}.pdf` ← Images de l'**ancrage**
|
||||||
|
3. `{seq}-{ref}.pdf` ← Fichier **Excel** converti
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## 🛠️ Configuration
|
||||||
|
|
||||||
|
Modifiez les constantes en haut du script :
|
||||||
|
|
||||||
|
```python
|
||||||
|
DEFAULT_BASE_DIR = Path.home() / "Projects" # Dossier racine des projets
|
||||||
|
DEFAULT_DEPART_DIR = Path.home() / "départ" # Dossier source
|
||||||
|
GEMINI_MODEL = "gemini-1.5-flash" # Modèle Gemini à utiliser
|
||||||
|
```
|
||||||
|
|
|
||||||
235
pole_automatio.py
Normal file
235
pole_automatio.py
Normal file
|
|
@ -0,0 +1,235 @@
|
||||||
|
import os
|
||||||
|
import re
|
||||||
|
import sys
|
||||||
|
import shutil
|
||||||
|
import base64
|
||||||
|
import argparse
|
||||||
|
import logging
|
||||||
|
import json
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
# -- Third-party --
|
||||||
|
try:
|
||||||
|
from PIL import Image
|
||||||
|
except ImportError:
|
||||||
|
print("❌ Pillow missing. Run: pip install pillow")
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
|
try:
|
||||||
|
import openpyxl
|
||||||
|
from openpyxl import load_workbook
|
||||||
|
except ImportError:
|
||||||
|
print("❌ openpyxl missing. Run: pip install openpyxl")
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
|
try:
|
||||||
|
from reportlab.lib.pagesizes import letter, landscape
|
||||||
|
from reportlab.platypus import SimpleDocTemplate, Table, TableStyle, Paragraph, Spacer
|
||||||
|
from reportlab.lib.styles import getSampleStyleSheet
|
||||||
|
from reportlab.lib import colors
|
||||||
|
except ImportError:
|
||||||
|
print("❌ reportlab missing. Run: pip install reportlab")
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
|
try:
|
||||||
|
import google.generativeai as genai
|
||||||
|
GEMINI_AVAILABLE = True
|
||||||
|
except ImportError:
|
||||||
|
GEMINI_AVAILABLE = False
|
||||||
|
|
||||||
|
# -- Logging --
|
||||||
|
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)-8s %(message)s", datefmt="%H:%M:%S")
|
||||||
|
log = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
DEFAULT_BASE_DIR = Path.home() /"PycharmProjects" / "structure" / "projet" # Where projects are created
|
||||||
|
DEFAULT_DEPART_DIR = Path.home() /"PycharmProjects" / "structure" / "départ" # Source folder with pole sub-folders
|
||||||
|
GEMINI_API_KEY = os.environ.get("AIzaSyDq2LmX_fwKGAwxGAtmBfX940vT2wDQzBU", "AIzaSyDq2LmX_fwKGAwxGAtmBfX940vT2wDQzBU") # Set via env var
|
||||||
|
GEMINI_MODEL = "models/gemini-2.5-flash"
|
||||||
|
|
||||||
|
FOLDER_PATTERN = re.compile(r"^(?P<seq>\d+)(?P<dero>sd|d)(?P<anchors>\d+)-(?P<pole_ref>[^\s]+)(?:\s+(?P<address>.+))?$", re.IGNORECASE)
|
||||||
|
|
||||||
|
def create_project_structure(project_name: str, base_dir: Path) -> Path:
|
||||||
|
project_root = base_dir / project_name
|
||||||
|
for folder in ["compilation", "ingénierie", "RSA-UDS", "plan"]:
|
||||||
|
(project_root / folder).mkdir(parents=True, exist_ok=True)
|
||||||
|
return project_root
|
||||||
|
|
||||||
|
def parse_pole_folder_name(folder_name: str) -> dict | None:
|
||||||
|
m = FOLDER_PATTERN.match(folder_name)
|
||||||
|
if not m: return None
|
||||||
|
return {
|
||||||
|
"seq": int(m.group("seq")),
|
||||||
|
"dero": m.group("dero").lower(),
|
||||||
|
"anchors": int(m.group("anchors")),
|
||||||
|
"pole_ref": m.group("pole_ref"),
|
||||||
|
"address": (m.group("address") or "").strip(),
|
||||||
|
"original": folder_name,
|
||||||
|
}
|
||||||
|
|
||||||
|
def compute_new_folder_name(parsed: dict, current_seq: int) -> str:
|
||||||
|
# Always starts with the sequence provided (1, 3, 5...)
|
||||||
|
if parsed["dero"] == "sd":
|
||||||
|
return f"{current_seq}0-{parsed['pole_ref']}"
|
||||||
|
else:
|
||||||
|
addr = f" {parsed['address']}" if parsed["address"] else ""
|
||||||
|
return f"{current_seq}1-{parsed['pole_ref']}{addr}"
|
||||||
|
|
||||||
|
def rename_excel_files(folder_path: Path, new_folder_name: str):
|
||||||
|
for f in folder_path.iterdir():
|
||||||
|
if f.suffix.lower() in (".xlsx", ".xls"):
|
||||||
|
# According to your request, UDS (Excel) should be named 10-
|
||||||
|
# We will handle the specific 10- prefix during the PDF conversion in process_pole_folder
|
||||||
|
pass
|
||||||
|
|
||||||
|
def identify_images_with_gemini(folder_path: Path) -> dict:
|
||||||
|
jpg_files = sorted(list(folder_path.glob("*.jpg")) + list(folder_path.glob("*.JPG")) + list(folder_path.glob("*.jpeg")))
|
||||||
|
if not jpg_files: return {"pole": [], "anchor": []}
|
||||||
|
|
||||||
|
if not GEMINI_AVAILABLE:
|
||||||
|
return _fallback_image_classification(jpg_files)
|
||||||
|
|
||||||
|
genai.configure(api_key=GEMINI_API_KEY)
|
||||||
|
model = genai.GenerativeModel(GEMINI_MODEL)
|
||||||
|
|
||||||
|
file_names = [f.name for f in jpg_files]
|
||||||
|
prompt = f"Classify these images as 'pole' or 'anchor'. Files: {file_names}. Respond ONLY in JSON."
|
||||||
|
|
||||||
|
content = [prompt]
|
||||||
|
for img_path in jpg_files:
|
||||||
|
with open(img_path, "rb") as f:
|
||||||
|
content.append({"mime_type": "image/jpeg", "data": base64.b64encode(f.read()).decode("utf-8")})
|
||||||
|
|
||||||
|
try:
|
||||||
|
response = model.generate_content(content)
|
||||||
|
raw = re.sub(r"^```json\s*|```$", "", response.text, flags=re.MULTILINE).strip()
|
||||||
|
classification = json.loads(raw)
|
||||||
|
result = {"pole": [], "anchor": []}
|
||||||
|
for fname, label in classification.items():
|
||||||
|
if label.lower() in result: result[label.lower()].append(fname)
|
||||||
|
return result
|
||||||
|
except:
|
||||||
|
return _fallback_image_classification(jpg_files)
|
||||||
|
|
||||||
|
def _fallback_image_classification(jpg_files):
|
||||||
|
mid = len(jpg_files) // 2
|
||||||
|
return {"pole": [f.name for f in jpg_files[:mid]], "anchor": [f.name for f in jpg_files[mid:]]}
|
||||||
|
|
||||||
|
#--------------------------------------------
|
||||||
|
# Transforme les photos en pdf
|
||||||
|
#--------------------------------------------
|
||||||
|
|
||||||
|
def images_to_pdf(image_paths: list[Path], output_pdf: Path):
|
||||||
|
pil_images = [Image.open(p).convert("RGB") for p in image_paths if p.exists()]
|
||||||
|
if pil_images:
|
||||||
|
pil_images[0].save(output_pdf, save_all=True, append_images=pil_images[1:])
|
||||||
|
|
||||||
|
def excel_to_pdf(excel_path: Path, output_pdf: Path):
|
||||||
|
wb = load_workbook(excel_path, data_only=True)
|
||||||
|
doc = SimpleDocTemplate(str(output_pdf), pagesize=landscape(letter))
|
||||||
|
styles = getSampleStyleSheet()
|
||||||
|
story = []
|
||||||
|
for sheet_name in wb.sheetnames:
|
||||||
|
ws = wb[sheet_name]
|
||||||
|
data = [[str(cell) if cell is not None else "" for cell in row] for row in ws.iter_rows(values_only=True)]
|
||||||
|
if not data: continue
|
||||||
|
t = Table(data, repeatRows=1)
|
||||||
|
t.setStyle(TableStyle([('FONTSIZE', (0,0), (-1,-1), 7), ('GRID', (0,0), (-1,-1), 0.5, colors.grey)]))
|
||||||
|
story.append(t)
|
||||||
|
doc.build(story)
|
||||||
|
|
||||||
|
def create_dummy_cdc(output_path: Path, label: str):
|
||||||
|
"""Creates a placeholder PDF for the CDC requirement."""
|
||||||
|
c = canvas.Canvas(str(output_path), pagesize=letter)
|
||||||
|
c.drawString(100, 750, f"CDC Document for {label}")
|
||||||
|
c.save()
|
||||||
|
|
||||||
|
#---------------------------
|
||||||
|
#This function is the "Orchestrator" for each individual pole.
|
||||||
|
#It decides the order of your documents and ensures the numbering matches your specific requirements.
|
||||||
|
#-----------------------------
|
||||||
|
|
||||||
|
def process_pole_folder(pole_folder: Path, compilation_dir: Path, folder_parsed: dict) -> int: # The path to the specific pole folder (e.g., 10-PBO345).
|
||||||
|
"""
|
||||||
|
Business Logic Order:
|
||||||
|
1. CDC (if 'd') -> prefix 10-
|
||||||
|
2. UDS (Excel) -> prefix 10- (or 11 if CDC exists)
|
||||||
|
3. Poteau (Images) -> prefix 11/12
|
||||||
|
4. Ancre (Images) -> prefix 12/13
|
||||||
|
"""
|
||||||
|
base_label = folder_parsed['pole_ref']
|
||||||
|
seq_prefix = pole_folder.name.split('-')[0] # The "10" or "11" etc
|
||||||
|
|
||||||
|
# 1. Identify Images
|
||||||
|
classification = identify_images_with_gemini(pole_folder)
|
||||||
|
|
||||||
|
# Logic for starting index based on your requirement "UDS must be 10-"
|
||||||
|
# If it's a 'd' folder, CDC is first.
|
||||||
|
current_pdf_idx = 10
|
||||||
|
|
||||||
|
# -- A. CDC (Calcul de Charge) - Only if 'd' --
|
||||||
|
if folder_parsed["dero"] == "d":
|
||||||
|
cdc_name = f"{current_pdf_idx}-{base_label} CDC.pdf"
|
||||||
|
|
||||||
|
# Logic to create or move CDC would go here. For now, we note the sequence shift.
|
||||||
|
log.info(f" 📑 CDC assigned to index {current_pdf_idx}")
|
||||||
|
current_pdf_idx += 1
|
||||||
|
|
||||||
|
# -- B. UDS (Excel) - Your request: Must be named 10- --
|
||||||
|
# (If CDC exists, this becomes 11-, if not, it stays 10-)
|
||||||
|
for xlsx_file in list(pole_folder.glob("*.xlsx")) + list(pole_folder.glob("*.xls")):
|
||||||
|
excel_pdf_name = f"{current_pdf_idx}-{base_label}.pdf"
|
||||||
|
print(excel_pdf_name)
|
||||||
|
excel_to_pdf(xlsx_file, compilation_dir / excel_pdf_name)
|
||||||
|
log.info(f" 📊 UDS Excel → {excel_pdf_name}")
|
||||||
|
current_pdf_idx += 1
|
||||||
|
|
||||||
|
# -- C. Poteau (Pole Images) --
|
||||||
|
pole_images = [pole_folder / n for n in classification["pole"]]
|
||||||
|
if pole_images:
|
||||||
|
pole_pdf_name = f"{current_pdf_idx}-{base_label}.pdf"
|
||||||
|
images_to_pdf(pole_images, compilation_dir / pole_pdf_name)
|
||||||
|
log.info(f" 🖼 Poteau → {pole_pdf_name}")
|
||||||
|
current_pdf_idx += 1
|
||||||
|
|
||||||
|
# -- D. Ancre (Anchor Images) --
|
||||||
|
anchor_images = [pole_folder / n for n in classification["anchor"]]
|
||||||
|
if anchor_images:
|
||||||
|
anchor_pdf_name = f"{current_pdf_idx}-{base_label}.pdf"
|
||||||
|
images_to_pdf(anchor_images, compilation_dir / anchor_pdf_name)
|
||||||
|
log.info(f" ⚓ Ancre → {anchor_pdf_name}")
|
||||||
|
current_pdf_idx += 1
|
||||||
|
|
||||||
|
return current_pdf_idx
|
||||||
|
|
||||||
|
def run(project_name: str, base_dir: Path, depart_dir: Path):
|
||||||
|
project_root = create_project_structure(project_name, base_dir)
|
||||||
|
rsa_uds_dir = project_root / "RSA-UDS"
|
||||||
|
compilation_dir = project_root / "compilation"
|
||||||
|
|
||||||
|
entries = []
|
||||||
|
for item in depart_dir.iterdir():
|
||||||
|
if item.is_dir():
|
||||||
|
p = parse_pole_folder_name(item.name)
|
||||||
|
if p: entries.append((p, item))
|
||||||
|
|
||||||
|
entries.sort(key=lambda x: x[0]["seq"])
|
||||||
|
|
||||||
|
current_folder_seq = 1
|
||||||
|
for parsed, src_path in entries:
|
||||||
|
new_name = compute_new_folder_name(parsed, current_folder_seq)
|
||||||
|
dst_path = rsa_uds_dir / new_name
|
||||||
|
if dst_path.exists(): shutil.rmtree(dst_path)
|
||||||
|
shutil.copytree(src_path, dst_path)
|
||||||
|
|
||||||
|
log.info(f"\n🔧 Processing: {new_name}")
|
||||||
|
process_pole_folder(dst_path, compilation_dir, parsed)
|
||||||
|
|
||||||
|
current_folder_seq += (2 if parsed["dero"] == "d" else 1)
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
parser = argparse.ArgumentParser()
|
||||||
|
parser.add_argument("project_name")
|
||||||
|
parser.add_argument("--base-dir", type=Path, default=DEFAULT_BASE_DIR)
|
||||||
|
parser.add_argument("--depart-dir", type=Path, default=DEFAULT_DEPART_DIR)
|
||||||
|
args = parser.parse_args()
|
||||||
|
run(args.project_name, args.base_dir, args.depart_dir)
|
||||||
243
pole_automation.py
Normal file
243
pole_automation.py
Normal file
|
|
@ -0,0 +1,243 @@
|
||||||
|
import os
|
||||||
|
import re
|
||||||
|
import sys
|
||||||
|
import shutil
|
||||||
|
import base64
|
||||||
|
import argparse
|
||||||
|
import logging
|
||||||
|
import json
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
# -- Third-party --
|
||||||
|
try:
|
||||||
|
from PIL import Image
|
||||||
|
except ImportError:
|
||||||
|
print("❌ Pillow missing. Run: pip install pillow")
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
|
try:
|
||||||
|
import openpyxl
|
||||||
|
from openpyxl import load_workbook
|
||||||
|
except ImportError:
|
||||||
|
print("❌ openpyxl missing. Run: pip install openpyxl")
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
|
try:
|
||||||
|
from reportlab.lib.pagesizes import letter, landscape
|
||||||
|
from reportlab.platypus import SimpleDocTemplate, Table, TableStyle, Paragraph, Spacer
|
||||||
|
from reportlab.lib.styles import getSampleStyleSheet
|
||||||
|
from reportlab.lib import colors
|
||||||
|
except ImportError:
|
||||||
|
print("❌ reportlab missing. Run: pip install reportlab")
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
|
try:
|
||||||
|
import google.generativeai as genai
|
||||||
|
GEMINI_AVAILABLE = True
|
||||||
|
except ImportError:
|
||||||
|
GEMINI_AVAILABLE = False
|
||||||
|
|
||||||
|
# -- Logging --
|
||||||
|
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)-8s %(message)s", datefmt="%H:%M:%S")
|
||||||
|
log = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
DEFAULT_BASE_DIR = Path.home() /"PycharmProjects" / "structure" / "projet" # Where projects are created
|
||||||
|
DEFAULT_DEPART_DIR = Path.home() /"PycharmProjects" / "structure" / "départ" # Source folder with pole sub-folders
|
||||||
|
GEMINI_API_KEY = os.environ.get("AIzaSyDq2LmX_fwKGAwxGAtmBfX940vT2wDQzBU", "AIzaSyDq2LmX_fwKGAwxGAtmBfX940vT2wDQzBU") # Set via env var
|
||||||
|
GEMINI_MODEL = "models/gemini-2.5-flash"
|
||||||
|
|
||||||
|
FOLDER_PATTERN = re.compile(r"^(?P<seq>\d+)(?P<dero>sd|d)(?P<anchors>\d+)-(?P<pole_ref>[^\s]+)(?:\s+(?P<address>.+))?$", re.IGNORECASE)
|
||||||
|
|
||||||
|
def create_project_structure(project_name: str, base_dir: Path) -> Path:
|
||||||
|
project_root = base_dir / project_name
|
||||||
|
for folder in ["compilation", "ingénierie", "RSA-UDS", "plan"]:
|
||||||
|
(project_root / folder).mkdir(parents=True, exist_ok=True)
|
||||||
|
return project_root
|
||||||
|
|
||||||
|
def parse_pole_folder_name(folder_name: str) -> dict | None:
|
||||||
|
m = FOLDER_PATTERN.match(folder_name)
|
||||||
|
if not m: return None
|
||||||
|
return {
|
||||||
|
"seq": int(m.group("seq")),
|
||||||
|
"dero": m.group("dero").lower(),
|
||||||
|
"anchors": int(m.group("anchors")),
|
||||||
|
"pole_ref": m.group("pole_ref"),
|
||||||
|
"address": (m.group("address") or "").strip(),
|
||||||
|
"original": folder_name,
|
||||||
|
}
|
||||||
|
|
||||||
|
def compute_new_folder_name(parsed: dict, current_seq: int) -> str:
|
||||||
|
# Always starts with the sequence provided (1, 3, 5...)
|
||||||
|
if parsed["dero"] == "sd":
|
||||||
|
return f"{current_seq}0-{parsed['pole_ref']}"
|
||||||
|
else:
|
||||||
|
addr = f" {parsed['address']}" if parsed["address"] else ""
|
||||||
|
return f"{current_seq}1-{parsed['pole_ref']}{addr}"
|
||||||
|
|
||||||
|
def rename_excel_files(folder_path: Path, new_folder_name: str):
|
||||||
|
for f in folder_path.iterdir():
|
||||||
|
if f.suffix.lower() in (".xlsx", ".xls"):
|
||||||
|
# According to your request, UDS (Excel) should be named 10-
|
||||||
|
# We will handle the specific 10- prefix during the PDF conversion in process_pole_folder
|
||||||
|
pass
|
||||||
|
|
||||||
|
def identify_images_with_gemini(folder_path: Path) -> dict:
|
||||||
|
jpg_files = sorted(list(folder_path.glob("*.jpg")) + list(folder_path.glob("*.JPG")) + list(folder_path.glob("*.jpeg")))
|
||||||
|
if not jpg_files: return {"pole": [], "anchor": []}
|
||||||
|
|
||||||
|
if not GEMINI_AVAILABLE:
|
||||||
|
return _fallback_image_classification(jpg_files)
|
||||||
|
|
||||||
|
genai.configure(api_key=GEMINI_API_KEY)
|
||||||
|
model = genai.GenerativeModel(GEMINI_MODEL)
|
||||||
|
|
||||||
|
file_names = [f.name for f in jpg_files]
|
||||||
|
prompt = f"Classify these images as 'pole' or 'anchor'. Files: {file_names}. Respond ONLY in JSON."
|
||||||
|
|
||||||
|
content = [prompt]
|
||||||
|
for img_path in jpg_files:
|
||||||
|
with open(img_path, "rb") as f:
|
||||||
|
content.append({"mime_type": "image/jpeg", "data": base64.b64encode(f.read()).decode("utf-8")})
|
||||||
|
|
||||||
|
try:
|
||||||
|
response = model.generate_content(content)
|
||||||
|
raw = re.sub(r"^```json\s*|```$", "", response.text, flags=re.MULTILINE).strip()
|
||||||
|
classification = json.loads(raw)
|
||||||
|
result = {"pole": [], "anchor": []}
|
||||||
|
for fname, label in classification.items():
|
||||||
|
if label.lower() in result: result[label.lower()].append(fname)
|
||||||
|
return result
|
||||||
|
except:
|
||||||
|
return _fallback_image_classification(jpg_files)
|
||||||
|
|
||||||
|
def _fallback_image_classification(jpg_files):
|
||||||
|
mid = len(jpg_files) // 2
|
||||||
|
return {"pole": [f.name for f in jpg_files[:mid]], "anchor": [f.name for f in jpg_files[mid:]]}
|
||||||
|
|
||||||
|
#--------------------------------------------
|
||||||
|
# Transforme les photos en pdf
|
||||||
|
#--------------------------------------------
|
||||||
|
|
||||||
|
def images_to_pdf(image_paths: list[Path], output_pdf: Path):
|
||||||
|
pil_images = [Image.open(p).convert("RGB") for p in image_paths if p.exists()]
|
||||||
|
if pil_images:
|
||||||
|
pil_images[0].save(output_pdf, save_all=True, append_images=pil_images[1:])
|
||||||
|
|
||||||
|
def excel_to_pdf(excel_path: Path, output_pdf: Path):
|
||||||
|
wb = load_workbook(excel_path, data_only=True)
|
||||||
|
doc = SimpleDocTemplate(str(output_pdf), pagesize=landscape(letter))
|
||||||
|
styles = getSampleStyleSheet()
|
||||||
|
story = []
|
||||||
|
for sheet_name in wb.sheetnames:
|
||||||
|
ws = wb[sheet_name]
|
||||||
|
data = [[str(cell) if cell is not None else "" for cell in row] for row in ws.iter_rows(values_only=True)]
|
||||||
|
if not data: continue
|
||||||
|
t = Table(data, repeatRows=1)
|
||||||
|
t.setStyle(TableStyle([('FONTSIZE', (0,0), (-1,-1), 7), ('GRID', (0,0), (-1,-1), 0.5, colors.grey)]))
|
||||||
|
story.append(t)
|
||||||
|
doc.build(story)
|
||||||
|
|
||||||
|
def create_dummy_cdc(output_path: Path, label: str):
|
||||||
|
"""Creates a placeholder PDF for the CDC requirement."""
|
||||||
|
c = canvas.Canvas(str(output_path), pagesize=letter)
|
||||||
|
c.drawString(100, 750, f"CDC Document for {label}")
|
||||||
|
c.save()
|
||||||
|
|
||||||
|
#---------------------------
|
||||||
|
#This function is the "Orchestrator" for each individual pole.
|
||||||
|
#It decides the order of your documents and ensures the numbering matches your specific requirements.
|
||||||
|
#-----------------------------
|
||||||
|
|
||||||
|
def process_pole_folder(pole_folder: Path, compilation_dir: Path, folder_parsed: dict, current_pdf_idx: int) -> int:
|
||||||
|
base_label = folder_parsed['pole_ref']
|
||||||
|
address = f" {folder_parsed['address']}" if folder_parsed['address'] else ""
|
||||||
|
full_label = f"{base_label}{address}" # Includes address if present (e.g. PHA627-3 Route 132)
|
||||||
|
|
||||||
|
# 1. Identify Images
|
||||||
|
classification = identify_images_with_gemini(pole_folder)
|
||||||
|
|
||||||
|
# -- A. CDC (Calcul de Charge) - Only if 'd' --
|
||||||
|
if folder_parsed["dero"] == "d":
|
||||||
|
cdc_name = f"{current_pdf_idx}-CDC-{full_label}.pdf"
|
||||||
|
# Logic to create/move CDC goes here
|
||||||
|
log.info(f" 📑 CDC assigned: {cdc_name}")
|
||||||
|
current_pdf_idx += 1
|
||||||
|
|
||||||
|
# -- B. UDS (Excel) -> Example: 10-RSA PHA627-1.pdf --
|
||||||
|
for xlsx_file in list(pole_folder.glob("*.xlsx")) + list(pole_folder.glob("*.xls")):
|
||||||
|
excel_pdf_name = f"{current_pdf_idx}-RSA {full_label}.pdf"
|
||||||
|
excel_to_pdf(xlsx_file, compilation_dir / excel_pdf_name)
|
||||||
|
log.info(f" 📊 UDS Excel → {excel_pdf_name}")
|
||||||
|
current_pdf_idx += 1
|
||||||
|
|
||||||
|
# -- C. Poteau (Pole Images) -> Example: 11-PHOTO PHA627-1.pdf --
|
||||||
|
pole_images = [pole_folder / n for n in classification["pole"]]
|
||||||
|
if pole_images:
|
||||||
|
pole_pdf_name = f"{current_pdf_idx}-PHOTO {full_label}.pdf"
|
||||||
|
images_to_pdf(pole_images, compilation_dir / pole_pdf_name)
|
||||||
|
log.info(f" 🖼 Poteau → {pole_pdf_name}")
|
||||||
|
current_pdf_idx += 1
|
||||||
|
|
||||||
|
# -- D. Ancre (Anchor Images) -> Example: 12-PHOTO ANCRE PHA627-1.pdf --
|
||||||
|
anchor_images = [pole_folder / n for n in classification["anchor"]]
|
||||||
|
if anchor_images:
|
||||||
|
anchor_pdf_name = f"{current_pdf_idx}-PHOTO ANCRE {full_label}.pdf"
|
||||||
|
images_to_pdf(anchor_images, compilation_dir / anchor_pdf_name)
|
||||||
|
log.info(f" ⚓ Ancre → {anchor_pdf_name}")
|
||||||
|
current_pdf_idx += 1
|
||||||
|
|
||||||
|
# We return the new index so the NEXT pole knows where to start
|
||||||
|
return current_pdf_idx
|
||||||
|
|
||||||
|
def run(project_name: str, base_dir: Path, depart_dir: Path):
|
||||||
|
# 1. Create the project structure
|
||||||
|
project_root = create_project_structure(project_name, base_dir)
|
||||||
|
rsa_uds_dir = project_root / "RSA-UDS"
|
||||||
|
compilation_dir = project_root / "compilation"
|
||||||
|
|
||||||
|
# --- ADDED: INITIALIZE ENTRIES ---
|
||||||
|
entries = []
|
||||||
|
if not depart_dir.exists():
|
||||||
|
log.error(f"❌ Source folder not found: {depart_dir}")
|
||||||
|
return
|
||||||
|
|
||||||
|
# Loop through the 'départ' folder to find pole sub-folders
|
||||||
|
for item in depart_dir.iterdir():
|
||||||
|
if item.is_dir():
|
||||||
|
parsed = parse_pole_folder_name(item.name)
|
||||||
|
if parsed:
|
||||||
|
entries.append((parsed, item))
|
||||||
|
else:
|
||||||
|
log.warning(f"⚠️ Skipping folder (unrecognized format): {item.name}")
|
||||||
|
# ---------------------------------
|
||||||
|
|
||||||
|
# 2. Sort folders to ensure 1sd1 comes before 2d2
|
||||||
|
entries.sort(key=lambda x: x[0]["seq"])
|
||||||
|
|
||||||
|
# GLOBAL PDF COUNTER (Starts at 10)
|
||||||
|
global_pdf_count = 10
|
||||||
|
|
||||||
|
for parsed, src_path in entries:
|
||||||
|
# 3. Create the RSA-UDS folder name
|
||||||
|
new_folder_name = compute_new_folder_name(parsed, parsed["seq"])
|
||||||
|
dst_path = rsa_uds_dir / new_folder_name
|
||||||
|
|
||||||
|
# Avoid error if folder already exists
|
||||||
|
if dst_path.exists():
|
||||||
|
shutil.rmtree(dst_path)
|
||||||
|
|
||||||
|
shutil.copytree(src_path, dst_path)
|
||||||
|
|
||||||
|
# 4. Process documents and update the global counter
|
||||||
|
log.info(f"\n🔧 Processing folder: {new_folder_name}")
|
||||||
|
global_pdf_count = process_pole_folder(dst_path, compilation_dir, parsed, global_pdf_count)
|
||||||
|
|
||||||
|
log.info(f"\n✅ All done! PDFs generated in: {compilation_dir}")
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
parser = argparse.ArgumentParser()
|
||||||
|
parser.add_argument("project_name")
|
||||||
|
parser.add_argument("--base-dir", type=Path, default=DEFAULT_BASE_DIR)
|
||||||
|
parser.add_argument("--depart-dir", type=Path, default=DEFAULT_DEPART_DIR)
|
||||||
|
args = parser.parse_args()
|
||||||
|
run(args.project_name, args.base_dir, args.depart_dir)
|
||||||
496
poteau.py
Normal file
496
poteau.py
Normal file
|
|
@ -0,0 +1,496 @@
|
||||||
|
"""
|
||||||
|
=============================================================================
|
||||||
|
POLE ENGINEERING PROJECT AUTOMATION SCRIPT
|
||||||
|
Automatisation des projets d'ingénierie de poteaux utilitaires
|
||||||
|
=============================================================================
|
||||||
|
Workflow:
|
||||||
|
1. Create project folder structure
|
||||||
|
2. Read source data from "départ" folder
|
||||||
|
3. Rename & move folders to RSA-UDS with business logic (sd/d)
|
||||||
|
4. Identify pole/anchor images via Gemini AI
|
||||||
|
5. Convert images & Excel files to PDF
|
||||||
|
6. Move all PDFs to compilation folder
|
||||||
|
=============================================================================
|
||||||
|
REQUIREMENTS:
|
||||||
|
pip install pillow openpyxl reportlab google-generativeai
|
||||||
|
=============================================================================
|
||||||
|
"""
|
||||||
|
|
||||||
|
import os
|
||||||
|
import re
|
||||||
|
import sys
|
||||||
|
import shutil
|
||||||
|
import base64
|
||||||
|
import argparse
|
||||||
|
import logging
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
# ── Third-party ──────────────────────────────────────────────────────────────
|
||||||
|
try:
|
||||||
|
from PIL import Image
|
||||||
|
except ImportError:
|
||||||
|
print("❌ Pillow missing. Run: pip install pillow")
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
|
try:
|
||||||
|
import openpyxl
|
||||||
|
from openpyxl import load_workbook
|
||||||
|
except ImportError:
|
||||||
|
print("❌ openpyxl missing. Run: pip install openpyxl")
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
|
try:
|
||||||
|
from reportlab.lib.pagesizes import letter, landscape
|
||||||
|
from reportlab.platypus import SimpleDocTemplate, Table, TableStyle, Paragraph, Spacer
|
||||||
|
from reportlab.lib.styles import getSampleStyleSheet
|
||||||
|
from reportlab.lib import colors
|
||||||
|
from reportlab.pdfgen import canvas
|
||||||
|
except ImportError:
|
||||||
|
print("❌ reportlab missing. Run: pip install reportlab")
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
|
try:
|
||||||
|
import google.generativeai as genai
|
||||||
|
GEMINI_AVAILABLE = True
|
||||||
|
except ImportError:
|
||||||
|
GEMINI_AVAILABLE = False
|
||||||
|
print("⚠️ google-generativeai not installed. Gemini AI disabled.")
|
||||||
|
print(" Run: pip install google-generativeai")
|
||||||
|
|
||||||
|
# ── Logging ───────────────────────────────────────────────────────────────────
|
||||||
|
logging.basicConfig(
|
||||||
|
level=logging.INFO,
|
||||||
|
format="%(asctime)s %(levelname)-8s %(message)s",
|
||||||
|
datefmt="%H:%M:%S",
|
||||||
|
)
|
||||||
|
log = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
# ═════════════════════════════════════════════════════════════════════════════
|
||||||
|
# CONFIGURATION – edit these paths to match your environment
|
||||||
|
# ═════════════════════════════════════════════════════════════════════════════
|
||||||
|
DEFAULT_BASE_DIR = Path.home() / "PycharmProjects" / "structure" / "projet" # Where projects are created
|
||||||
|
DEFAULT_DEPART_DIR = Path.home() / "PycharmProjects" / "structure" / "départ" # Source folder with pole sub-folders
|
||||||
|
GEMINI_API_KEY = os.environ.get("AIzaSyDq2LmX_fwKGAwxGAtmBfX940vT2wDQzBU") # Set via env var
|
||||||
|
GEMINI_MODEL = "models/gemini-2.5-flash"
|
||||||
|
|
||||||
|
|
||||||
|
# ═════════════════════════════════════════════════════════════════════════════
|
||||||
|
# STEP 1 – PROJECT FOLDER CREATION
|
||||||
|
# ═════════════════════════════════════════════════════════════════════════════
|
||||||
|
|
||||||
|
SUB_FOLDERS = ["compilation", "ingénierie", "RSA-UDS", "plan"]
|
||||||
|
|
||||||
|
def create_project_structure(project_name: str, base_dir: Path) -> Path:
|
||||||
|
"""
|
||||||
|
Create:
|
||||||
|
<base_dir>/<project_name>/
|
||||||
|
compilation/
|
||||||
|
ingénierie/
|
||||||
|
RSA-UDS/
|
||||||
|
plan/
|
||||||
|
Returns the project root path.
|
||||||
|
"""
|
||||||
|
project_root = base_dir / project_name
|
||||||
|
project_root.mkdir(parents=True, exist_ok=True)
|
||||||
|
|
||||||
|
for folder in SUB_FOLDERS:
|
||||||
|
(project_root / folder).mkdir(exist_ok=True)
|
||||||
|
|
||||||
|
log.info(f"✅ Project structure created: {project_root}")
|
||||||
|
return project_root
|
||||||
|
|
||||||
|
|
||||||
|
# ═════════════════════════════════════════════════════════════════════════════
|
||||||
|
# STEP 2 – PARSE SOURCE POLE FOLDERS
|
||||||
|
# ═════════════════════════════════════════════════════════════════════════════
|
||||||
|
|
||||||
|
# Pattern: <seq><d|sd><anchors>-<pole_ref> <address>
|
||||||
|
# e.g. 1d1-PBO345 rue saint-eugene
|
||||||
|
# 2sd2-PBO678 boul. des érables
|
||||||
|
FOLDER_PATTERN = re.compile(
|
||||||
|
r"^(?P<seq>\d+)"
|
||||||
|
r"(?P<dero>sd|d)"
|
||||||
|
r"(?P<anchors>\d+)"
|
||||||
|
r"-"
|
||||||
|
r"(?P<pole_ref>[^\s]+)"
|
||||||
|
r"(?:\s+(?P<address>.+))?$",
|
||||||
|
re.IGNORECASE,
|
||||||
|
)
|
||||||
|
|
||||||
|
def parse_pole_folder_name(folder_name: str) -> dict | None:
|
||||||
|
"""
|
||||||
|
Parse a source folder name and return a dict with extracted components,
|
||||||
|
or None if the name doesn't match the expected pattern.
|
||||||
|
"""
|
||||||
|
m = FOLDER_PATTERN.match(folder_name)
|
||||||
|
if not m:
|
||||||
|
return None
|
||||||
|
return {
|
||||||
|
"seq" : int(m.group("seq")),
|
||||||
|
"dero" : m.group("dero").lower(), # 'sd' or 'd'
|
||||||
|
"anchors" : int(m.group("anchors")),
|
||||||
|
"pole_ref" : m.group("pole_ref"), # e.g. PBO345
|
||||||
|
"address" : (m.group("address") or "").strip(),
|
||||||
|
"original" : folder_name,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
# ═════════════════════════════════════════════════════════════════════════════
|
||||||
|
# STEP 3 – RENAMING & MOVING LOGIC (sd vs d)
|
||||||
|
# ═════════════════════════════════════════════════════════════════════════════
|
||||||
|
|
||||||
|
def compute_new_folder_name(parsed: dict, current_seq: int) -> str:
|
||||||
|
"""
|
||||||
|
Apply business naming rules:
|
||||||
|
sd → 10-PBO345 (prefix = current_seq, no address)
|
||||||
|
d → 11-PBO345 rue saint-x (prefix = current_seq, with address)
|
||||||
|
|
||||||
|
NOTE: The leading '1' before 0/1 comes from your spec (10- / 11-).
|
||||||
|
If you want the sequence itself used as prefix, adjust here.
|
||||||
|
"""
|
||||||
|
if parsed["dero"] == "sd":
|
||||||
|
return f"{current_seq}0-{parsed['pole_ref']}"
|
||||||
|
else:
|
||||||
|
addr = f" {parsed['address']}" if parsed["address"] else ""
|
||||||
|
return f"{current_seq}1-{parsed['pole_ref']}{addr}"
|
||||||
|
|
||||||
|
|
||||||
|
def rename_excel_files(folder_path: Path, new_folder_name: str):
|
||||||
|
"""Rename any .xlsx / .xls inside folder to match new_folder_name."""
|
||||||
|
for f in folder_path.iterdir():
|
||||||
|
if f.suffix.lower() in (".xlsx", ".xls"):
|
||||||
|
new_name = new_folder_name + f.suffix
|
||||||
|
f.rename(folder_path / new_name)
|
||||||
|
log.info(f" 📄 Excel renamed → {new_name}")
|
||||||
|
|
||||||
|
|
||||||
|
def process_depart_to_rsa_uds(depart_dir: Path, rsa_uds_dir: Path) -> list[Path]:
|
||||||
|
"""
|
||||||
|
Read all sub-folders in départ, sort by sequence number,
|
||||||
|
apply renaming logic, move to RSA-UDS.
|
||||||
|
Returns list of destination folder paths (for further processing).
|
||||||
|
"""
|
||||||
|
entries = []
|
||||||
|
for item in depart_dir.iterdir():
|
||||||
|
if item.is_dir():
|
||||||
|
parsed = parse_pole_folder_name(item.name)
|
||||||
|
if parsed:
|
||||||
|
entries.append((parsed, item))
|
||||||
|
else:
|
||||||
|
log.warning(f"⚠️ Unrecognised folder name (skipped): {item.name}")
|
||||||
|
|
||||||
|
# Sort by original sequence
|
||||||
|
entries.sort(key=lambda x: x[0]["seq"])
|
||||||
|
|
||||||
|
current_seq = 1
|
||||||
|
moved_folders = []
|
||||||
|
|
||||||
|
for parsed, src_path in entries:
|
||||||
|
new_name = compute_new_folder_name(parsed, current_seq)
|
||||||
|
dst_path = rsa_uds_dir / new_name
|
||||||
|
|
||||||
|
shutil.copytree(src_path, dst_path)
|
||||||
|
log.info(f"📁 {src_path.name!r:40s} → {new_name!r}")
|
||||||
|
|
||||||
|
# Rename Excel files inside the destination folder
|
||||||
|
rename_excel_files(dst_path, new_name)
|
||||||
|
|
||||||
|
moved_folders.append(dst_path)
|
||||||
|
|
||||||
|
# Advance counter
|
||||||
|
if parsed["dero"] == "sd":
|
||||||
|
current_seq += 1
|
||||||
|
else:
|
||||||
|
current_seq += 2 # Extra jump for CDC document
|
||||||
|
|
||||||
|
return moved_folders
|
||||||
|
|
||||||
|
|
||||||
|
# ═════════════════════════════════════════════════════════════════════════════
|
||||||
|
# STEP 4 – GEMINI AI IMAGE IDENTIFICATION
|
||||||
|
# ═════════════════════════════════════════════════════════════════════════════
|
||||||
|
|
||||||
|
def _encode_image(image_path: Path) -> str:
|
||||||
|
"""Return base64-encoded image content."""
|
||||||
|
with open(image_path, "rb") as f:
|
||||||
|
return base64.b64encode(f.read()).decode("utf-8")
|
||||||
|
|
||||||
|
|
||||||
|
def identify_images_with_gemini(folder_path: Path) -> dict:
|
||||||
|
"""
|
||||||
|
Send all JPG images in folder to Gemini and ask it to label each as
|
||||||
|
'pole' or 'anchor'.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
{
|
||||||
|
"pole": ["IMG_001.jpg", ...],
|
||||||
|
"anchor": ["IMG_002.jpg", ...],
|
||||||
|
"unknown": [...]
|
||||||
|
}
|
||||||
|
"""
|
||||||
|
jpg_files = sorted(folder_path.glob("*.jpg")) + sorted(folder_path.glob("*.JPG")) + \
|
||||||
|
sorted(folder_path.glob("*.jpeg"))
|
||||||
|
|
||||||
|
if not jpg_files:
|
||||||
|
log.warning(f" No JPG images found in {folder_path.name}")
|
||||||
|
return {"pole": [], "anchor": [], "unknown": []}
|
||||||
|
|
||||||
|
if not GEMINI_AVAILABLE or not GEMINI_API_KEY:
|
||||||
|
log.warning(" Gemini unavailable – using filename heuristic fallback.")
|
||||||
|
return _fallback_image_classification(jpg_files)
|
||||||
|
|
||||||
|
genai.configure(api_key=GEMINI_API_KEY)
|
||||||
|
model = genai.GenerativeModel(GEMINI_MODEL)
|
||||||
|
|
||||||
|
# Build the content list
|
||||||
|
file_names = [f.name for f in jpg_files]
|
||||||
|
prompt = (
|
||||||
|
"You are analysing images for a utility pole engineering project. "
|
||||||
|
"For each image provided, classify it as either 'pole' or 'anchor'. "
|
||||||
|
"A pole image shows a utility/telephone pole. "
|
||||||
|
"An anchor image shows a ground anchor or guy-wire anchor attachment. "
|
||||||
|
f"The filenames are: {file_names}. "
|
||||||
|
"Respond ONLY with a JSON object where each key is a filename and "
|
||||||
|
"the value is either 'pole' or 'anchor'. Example: "
|
||||||
|
'{"IMG_001.jpg": "pole", "IMG_002.jpg": "anchor"}'
|
||||||
|
)
|
||||||
|
|
||||||
|
content = [prompt]
|
||||||
|
for img_path in jpg_files:
|
||||||
|
content.append({
|
||||||
|
"mime_type": "image/jpeg",
|
||||||
|
"data": _encode_image(img_path),
|
||||||
|
})
|
||||||
|
|
||||||
|
try:
|
||||||
|
response = model.generate_content(content)
|
||||||
|
raw = response.text.strip()
|
||||||
|
# Strip markdown code fences if present
|
||||||
|
raw = re.sub(r"^```json\s*|```$", "", raw, flags=re.MULTILINE).strip()
|
||||||
|
import json
|
||||||
|
classification = json.loads(raw)
|
||||||
|
except Exception as e:
|
||||||
|
log.error(f" Gemini error: {e} – falling back to heuristic.")
|
||||||
|
return _fallback_image_classification(jpg_files)
|
||||||
|
|
||||||
|
result = {"pole": [], "anchor": [], "unknown": []}
|
||||||
|
for fname, label in classification.items():
|
||||||
|
label = label.lower()
|
||||||
|
if label in result:
|
||||||
|
result[label].append(fname)
|
||||||
|
else:
|
||||||
|
result["unknown"].append(fname)
|
||||||
|
|
||||||
|
log.info(f" 🤖 Gemini classified: pole={result['pole']}, anchor={result['anchor']}")
|
||||||
|
return result
|
||||||
|
|
||||||
|
|
||||||
|
def _fallback_image_classification(jpg_files: list[Path]) -> dict:
|
||||||
|
"""
|
||||||
|
Simple heuristic fallback when Gemini is unavailable.
|
||||||
|
Files with 'pole' or 'p' in name → pole; 'anchor' or 'a' → anchor.
|
||||||
|
Remaining: first half = pole, second half = anchor.
|
||||||
|
"""
|
||||||
|
result = {"pole": [], "anchor": [], "unknown": []}
|
||||||
|
unclassified = []
|
||||||
|
|
||||||
|
for f in jpg_files:
|
||||||
|
lower = f.name.lower()
|
||||||
|
if "pole" in lower or lower.startswith("p"):
|
||||||
|
result["pole"].append(f.name)
|
||||||
|
elif "anchor" in lower or lower.startswith("a"):
|
||||||
|
result["anchor"].append(f.name)
|
||||||
|
else:
|
||||||
|
unclassified.append(f.name)
|
||||||
|
|
||||||
|
mid = len(unclassified) // 2
|
||||||
|
result["pole"].extend(unclassified[:mid])
|
||||||
|
result["anchor"].extend(unclassified[mid:])
|
||||||
|
|
||||||
|
return result
|
||||||
|
|
||||||
|
|
||||||
|
# ═════════════════════════════════════════════════════════════════════════════
|
||||||
|
# STEP 5 – FILE CONVERSION & FINAL SORTING
|
||||||
|
# ═════════════════════════════════════════════════════════════════════════════
|
||||||
|
|
||||||
|
def images_to_pdf(image_paths: list[Path], output_pdf: Path):
|
||||||
|
"""Combine one or more JPG images into a single PDF page-per-image."""
|
||||||
|
if not image_paths:
|
||||||
|
return
|
||||||
|
|
||||||
|
pil_images = []
|
||||||
|
for p in image_paths:
|
||||||
|
img = Image.open(p).convert("RGB")
|
||||||
|
pil_images.append(img)
|
||||||
|
|
||||||
|
if pil_images:
|
||||||
|
first = pil_images[0]
|
||||||
|
rest = pil_images[1:]
|
||||||
|
first.save(output_pdf, save_all=True, append_images=rest)
|
||||||
|
log.info(f" 🖼→📄 {[p.name for p in image_paths]} → {output_pdf.name}")
|
||||||
|
|
||||||
|
|
||||||
|
def excel_to_pdf(excel_path: Path, output_pdf: Path):
|
||||||
|
"""
|
||||||
|
Convert an Excel file to a PDF using ReportLab.
|
||||||
|
Reads all sheets and renders them as tables.
|
||||||
|
"""
|
||||||
|
wb = load_workbook(excel_path, data_only=True)
|
||||||
|
doc = SimpleDocTemplate(
|
||||||
|
str(output_pdf),
|
||||||
|
pagesize=landscape(letter),
|
||||||
|
rightMargin=20, leftMargin=20,
|
||||||
|
topMargin=30, bottomMargin=20,
|
||||||
|
)
|
||||||
|
styles = getSampleStyleSheet()
|
||||||
|
story = []
|
||||||
|
|
||||||
|
for sheet_name in wb.sheetnames:
|
||||||
|
ws = wb[sheet_name]
|
||||||
|
story.append(Paragraph(f"Sheet: {sheet_name}", styles["Heading2"]))
|
||||||
|
story.append(Spacer(1, 6))
|
||||||
|
|
||||||
|
data = []
|
||||||
|
for row in ws.iter_rows(values_only=True):
|
||||||
|
data.append([str(cell) if cell is not None else "" for cell in row])
|
||||||
|
|
||||||
|
if not data:
|
||||||
|
continue
|
||||||
|
|
||||||
|
col_count = max(len(r) for r in data)
|
||||||
|
col_width = (landscape(letter)[0] - 40) / max(col_count, 1)
|
||||||
|
|
||||||
|
table = Table(data, colWidths=[col_width] * col_count, repeatRows=1)
|
||||||
|
table.setStyle(TableStyle([
|
||||||
|
("BACKGROUND", (0, 0), (-1, 0), colors.HexColor("#2C4770")),
|
||||||
|
("TEXTCOLOR", (0, 0), (-1, 0), colors.white),
|
||||||
|
("FONTSIZE", (0, 0), (-1, -1), 7),
|
||||||
|
("GRID", (0, 0), (-1, -1), 0.25, colors.grey),
|
||||||
|
("ROWBACKGROUNDS", (0, 1), (-1, -1), [colors.white, colors.HexColor("#F0F4FA")]),
|
||||||
|
("VALIGN", (0, 0), (-1, -1), "MIDDLE"),
|
||||||
|
]))
|
||||||
|
story.append(table)
|
||||||
|
story.append(Spacer(1, 20))
|
||||||
|
|
||||||
|
doc.build(story)
|
||||||
|
log.info(f" 📊→📄 {excel_path.name} → {output_pdf.name}")
|
||||||
|
|
||||||
|
|
||||||
|
def process_pole_folder(
|
||||||
|
pole_folder: Path,
|
||||||
|
compilation_dir: Path,
|
||||||
|
pdf_seq: int,
|
||||||
|
) -> int:
|
||||||
|
"""
|
||||||
|
For a single pole folder (already in RSA-UDS):
|
||||||
|
1. Identify images via Gemini
|
||||||
|
2. Convert pole images → PDF (pdf_seq)
|
||||||
|
3. Convert anchor images → PDF (pdf_seq + 1)
|
||||||
|
4. Convert Excel → PDF (pdf_seq + 2)
|
||||||
|
5. Move all PDFs to compilation
|
||||||
|
|
||||||
|
Returns the next available pdf_seq.
|
||||||
|
"""
|
||||||
|
folder_label = pole_folder.name # e.g. "10-PBO345"
|
||||||
|
|
||||||
|
# Strip leading numeric prefix for base label e.g. "PBO345"
|
||||||
|
base_label = re.sub(r"^\d+-", "", folder_label)
|
||||||
|
|
||||||
|
# ── Identify images ──────────────────────────────────────────────────────
|
||||||
|
classification = identify_images_with_gemini(pole_folder)
|
||||||
|
|
||||||
|
# ── Pole PDF ─────────────────────────────────────────────────────────────
|
||||||
|
pole_images = [pole_folder / n for n in classification["pole"] if (pole_folder / n).exists()]
|
||||||
|
if pole_images:
|
||||||
|
pole_pdf_name = f"{pdf_seq}-{base_label}.pdf"
|
||||||
|
images_to_pdf(pole_images, compilation_dir / pole_pdf_name)
|
||||||
|
pdf_seq += 1
|
||||||
|
|
||||||
|
# ── Anchor PDF ───────────────────────────────────────────────────────────
|
||||||
|
anchor_images = [pole_folder / n for n in classification["anchor"] if (pole_folder / n).exists()]
|
||||||
|
if anchor_images:
|
||||||
|
anchor_pdf_name = f"{pdf_seq}-{base_label}.pdf"
|
||||||
|
images_to_pdf(anchor_images, compilation_dir / anchor_pdf_name)
|
||||||
|
pdf_seq += 1
|
||||||
|
|
||||||
|
# ── Excel → PDF ──────────────────────────────────────────────────────────
|
||||||
|
for xlsx_file in list(pole_folder.glob("*.xlsx")) + list(pole_folder.glob("*.xls")):
|
||||||
|
excel_pdf_name = f"{pdf_seq}-{base_label}.pdf"
|
||||||
|
excel_to_pdf(xlsx_file, compilation_dir / excel_pdf_name)
|
||||||
|
pdf_seq += 1
|
||||||
|
|
||||||
|
return pdf_seq
|
||||||
|
|
||||||
|
|
||||||
|
# ═════════════════════════════════════════════════════════════════════════════
|
||||||
|
# MAIN ORCHESTRATOR
|
||||||
|
# ═════════════════════════════════════════════════════════════════════════════
|
||||||
|
|
||||||
|
def run(project_name: str, base_dir: Path, depart_dir: Path):
|
||||||
|
log.info("=" * 65)
|
||||||
|
log.info(f" PROJECT : {project_name}")
|
||||||
|
log.info(f" BASE DIR: {base_dir}")
|
||||||
|
log.info(f" DÉPART : {depart_dir}")
|
||||||
|
log.info("=" * 65)
|
||||||
|
|
||||||
|
# 1 – Create structure
|
||||||
|
project_root = create_project_structure(project_name, base_dir)
|
||||||
|
rsa_uds_dir = project_root / "RSA-UDS"
|
||||||
|
compilation_dir = project_root / "compilation"
|
||||||
|
|
||||||
|
# 2 & 3 – Move and rename pole folders
|
||||||
|
if not depart_dir.exists():
|
||||||
|
log.error(f"Source départ folder not found: {depart_dir}")
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
|
moved_folders = process_depart_to_rsa_uds(depart_dir, rsa_uds_dir)
|
||||||
|
log.info(f"📂 {len(moved_folders)} pole folder(s) moved to RSA-UDS.")
|
||||||
|
|
||||||
|
# 4 & 5 – Process each pole folder
|
||||||
|
pdf_seq = 1
|
||||||
|
for folder in sorted(moved_folders):
|
||||||
|
log.info(f"\n🔧 Processing: {folder.name}")
|
||||||
|
pdf_seq = process_pole_folder(folder, compilation_dir, pdf_seq)
|
||||||
|
|
||||||
|
log.info("\n" + "=" * 65)
|
||||||
|
log.info(f"✅ DONE. All PDFs are in: {compilation_dir}")
|
||||||
|
log.info("=" * 65)
|
||||||
|
|
||||||
|
|
||||||
|
# ═════════════════════════════════════════════════════════════════════════════
|
||||||
|
# CLI ENTRY POINT
|
||||||
|
# ═════════════════════════════════════════════════════════════════════════════
|
||||||
|
|
||||||
|
def main():
|
||||||
|
parser = argparse.ArgumentParser(
|
||||||
|
description="Utility pole engineering project automation script",
|
||||||
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||||||
|
epilog="""
|
||||||
|
Examples:
|
||||||
|
python pole_automation.py E65-BK7-26-132679
|
||||||
|
python pole_automation.py E65-BK7-26-132679 --base-dir /Users/me/Projects
|
||||||
|
python pole_automation.py E65-BK7-26-132679 --depart-dir /Users/me/départ
|
||||||
|
python pole_automation.py E65-BK7-26-132679 --gemini-key AIza...
|
||||||
|
""",
|
||||||
|
)
|
||||||
|
parser.add_argument("project_name", help="Project name (e.g. E65-BK7-26-132679)")
|
||||||
|
parser.add_argument("--base-dir", type=Path, default=DEFAULT_BASE_DIR,
|
||||||
|
help=f"Root directory for projects (default: {DEFAULT_BASE_DIR})")
|
||||||
|
parser.add_argument("--depart-dir", type=Path, default=DEFAULT_DEPART_DIR,
|
||||||
|
help=f"Source 'départ' folder (default: {DEFAULT_DEPART_DIR})")
|
||||||
|
parser.add_argument("--gemini-key", type=str, default=GEMINI_API_KEY,
|
||||||
|
help="Gemini API key (or set GEMINI_API_KEY env var)")
|
||||||
|
|
||||||
|
args = parser.parse_args()
|
||||||
|
|
||||||
|
# Inject Gemini key if provided via CLI
|
||||||
|
if args.gemini_key:
|
||||||
|
global GEMINI_API_KEY
|
||||||
|
GEMINI_API_KEY = args.gemini_key
|
||||||
|
|
||||||
|
run(args.project_name, args.base_dir, args.depart_dir)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
main()
|
||||||
Loading…
Reference in New Issue
Block a user