bede/tfm_ainventory
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

blabla

Browse Source
This commit is contained in:
Daniel Bedeleanu
2026-04-14 22:54:19 +03:00
parent 1ea5e90bc4
commit 6c57b1b0c2
69 changed files with 1020 additions and 554 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
backend/ai/gemini.py
View File

@@ -1,42 +1,50 @@
import os
import json
import io
import logging
from PIL import Image
from google import genai
from google.genai import types
log = logging.getLogger("ainventory")
def get_best_models():
# Using the exact models discovered via diagnostic
return ["gemini-2.0-flash", "gemini-2.5-flash"]
# Priority list based on aliases and future-gen models found in user logs
return [
"gemini-flash-latest", # Points to the best stable Flash version
"gemini-3.1-flash-lite-preview", # Cutting edge 3.1 Lite
"gemini-flash-lite-latest", # Best Lite alias
"gemini-pro-latest" # Pro fallback alias
]
def extract(image_bytes: bytes, prompt: str):
api_key = os.environ.get("GEMINI_API_KEY")
if not api_key:
print("CRITICAL: GEMINI_API_KEY is MISSING in environment!")
log.error("CRITICAL: GEMINI_API_KEY is MISSING in environment!")
return None
# Log partial key for safety debug
key_hint = f"{api_key[:4]}...{api_key[-4:]}" if len(api_key) > 8 else "too short"
print(f"🔑 Using API Key: {key_hint}")
log.info(f"🔑 Using API Key: {key_hint}")
try:
# Initialize the NEW SDK Client forcing stable v1 API
client = genai.Client(api_key=api_key, http_options={'api_version': 'v1'})
# Switching to v1beta which is required for many experimental/new models
client = genai.Client(api_key=api_key, http_options={'api_version': 'v1beta'})
# DEBUG: List allowed models for this key
print("🔍 Checking available models for your key...")
log.debug("🔍 Checking available models for your key...")
try:
for m in client.models.list():
print(f" - Found: {m.name}")
log.debug(f" - Found: {m.name}")
except Exception as list_e:
print(f" ⚠️ Could not list models: {list_e}")
log.warning(f" ⚠️ Could not list models: {list_e}")
models_to_try = get_best_models()
# Try models in order
for model_name in models_to_try:
try:
print(f"🚀 AI Launching (v2 SDK): {model_name}...")
log.info(f"🚀 AI Launching (v2 SDK): {model_name}...")
# In the new SDK, we pass a list of parts (text string and image bytes)
response = client.models.generate_content(
@@ -48,10 +56,12 @@ def extract(image_bytes: bytes, prompt: str):
)
if not response or not response.text:
log.warning(f"⚠️ Gemini {model_name} returned NO text.")
continue
text = response.text.strip()
print(f"✅ AI Response Received ({len(text)} bytes)")
log.info(f"✅ AI Response Received ({len(text)} bytes)")
log.debug(f"FULL RESPONSE:\n{text}")
# Extract JSON block
if "```json" in text:
@@ -62,10 +72,10 @@ def extract(image_bytes: bytes, prompt: str):
return json.loads(text)
except Exception as e:
print(f"❌ Gemini {model_name} failed: {e}")
log.error(f"❌ Gemini {model_name} failed: {e}")
continue
except Exception as outer_e:
print(f"❌ Gemini Client Init failed: {outer_e}")
log.error(f"❌ Gemini Client Init failed: {outer_e}")
return None

104
backend/ai_vision.py
View File

@@ -1,10 +1,42 @@
import os
import time
from dotenv import load_dotenv
from . import models
from .database import SessionLocal
from .ai import gemini, claude
# Load environment variables from the directory where this file resides
# Note: Environment variables are managed centrally by config_loader.py
base_dir = os.path.dirname(os.path.abspath(__file__))
dotenv_path = os.path.join(base_dir, ".env")
load_dotenv(dotenv_path)
class PromptManager:
"""Manages AI prompt with auto-reloading from file."""
def __init__(self, file_path: str):
self.file_path = file_path
self.last_mtime = 0
self.cached_prompt = None
def get_prompt(self) -> str:
if not os.path.exists(self.file_path):
return None
try:
current_mtime = os.path.getmtime(self.file_path)
if current_mtime > self.last_mtime or self.cached_prompt is None:
with open(self.file_path, 'r', encoding='utf-8') as f:
self.cached_prompt = f.read().strip()
self.last_mtime = current_mtime
# Use print or log for visibility in dev
print(f"🔄 AI Vision prompt reloaded from {self.file_path} (mtime: {current_mtime})")
return self.cached_prompt
except Exception as e:
print(f"⚠️ Failed to reload AI prompt: {e}")
return self.cached_prompt
# The prompt file is located in the global /config directory
# We go up one level from backend/ to reach project root, then into config/
PROJECT_ROOT = os.path.dirname(os.path.abspath(base_dir))
PROMPT_FILE_PATH = os.path.join(PROJECT_ROOT, "config", "ai_prompt.md")
prompt_mgr = PromptManager(PROMPT_FILE_PATH)
def extract_label_info(image_bytes: bytes, mode: str = "item"):
"""
@@ -29,20 +61,27 @@ def extract_label_info(image_bytes: bytes, mode: str = "item"):
}
"""
else:
# Fetch custom prompt from DB
setting = db.query(models.SystemSetting).filter(models.SystemSetting.key == "ai_extraction_prompt").first()
if setting:
prompt = setting.value
else:
# Fallback to a sensible default if DB is not ready
prompt = "Extract technical specs. Return JSON with name, category, description, connector, size, color, part_number, ocr_text, quantity."
# 1. Try fetching from the configuration file first (SSOT)
prompt = prompt_mgr.get_prompt()
if not prompt:
# 2. Fallback to Database if file is missing
setting = db.query(models.SystemSetting).filter(models.SystemSetting.key == "ai_extraction_prompt").first()
if setting:
prompt = setting.value
else:
# 3. Final fallback to hardcoded default
prompt = "Extract technical specs. Return JSON with name, category, description, connector, size, color, part_number, ocr_text, quantity."
# 1. Try Gemini
result = gemini.extract(image_bytes, prompt)
if result:
# Map user-defined prompt keys to model fields if needed
# User keys: Item, Type, Description, Category, Connector, Size, Color, PartNr, OCR
# Check if AI returned a list of items or a singular object
raw_items = result.get("items") or result.get("Items")
was_list = isinstance(raw_items, list)
items_to_map = raw_items if was_list else [result]
mapping = {
"Item": "name",
"Type": "type",
@@ -55,23 +94,32 @@ def extract_label_info(image_bytes: bytes, mode: str = "item"):
"OCR": "ocr_text"
}
final_result = {}
for ai_key, model_key in mapping.items():
if ai_key in result:
final_result[model_key] = result[ai_key]
elif model_key in result: # Already mapped or using model keys
final_result[model_key] = result[model_key]
mapped_items = []
for item_data in items_to_map:
final_item = {}
for ai_key, model_key in mapping.items():
val = item_data.get(ai_key) or item_data.get(model_key)
if val and isinstance(val, str):
final_item[model_key] = val.strip()
else:
final_item[model_key] = val
# Default fields
final_item["quantity"] = item_data.get("quantity", 1)
raw_barcode = item_data.get("barcode") or item_data.get("PartNr") or item_data.get("part_number") or item_data.get("Part Number")
final_item["barcode"] = str(raw_barcode).strip() if raw_barcode else f"AI-{int(time.time()*100)}"
# Handle Box mode specifically inside mapping
if mode == "box":
final_item["box_label"] = final_item.get("box_label") or item_data.get("Box") or final_item.get("name") or "Unknown Box"
final_item["name"] = final_item["box_label"]
mapped_items.append(final_item)
# Ensure quantity and barcode are handled if returned or default
final_result["quantity"] = result.get("quantity", 1)
final_result["barcode"] = result.get("barcode", result.get("PartNr", result.get("part_number", "")))
# Handle Box mode specifically
if mode == "box":
final_result["box_label"] = result.get("box_label", result.get("name", "Unknown Box"))
final_result["name"] = final_result["box_label"]
return final_result
# Return either the whole list wrapper or the first item (legacy compatibility)
if was_list:
return {"items": mapped_items}
return mapped_items[0] if mapped_items else {"error": "No items after mapping"}
finally:
db.close()

34
backend/config_loader.py Normal file
View File

@@ -0,0 +1,34 @@
import os
import logging
from dotenv import load_dotenv
log = logging.getLogger("ainventory")
def load_config():
"""
Centralized environment loader for TFM aInventory.
Prioritizes existing environment variables (Docker),
then inventory.env at project root, then backend/.env.
"""
# Base directory is backend/
base_dir = os.path.dirname(os.path.abspath(__file__))
# Project root is one level up
project_root = os.path.dirname(base_dir)
inventory_env_path = os.path.join(project_root, "inventory.env")
backend_env_path = os.path.join(base_dir, ".env")
# Check for inventory.env in root (Master Config)
if os.path.exists(inventory_env_path):
load_dotenv(inventory_env_path)
log.info(f"✅ Loaded master configuration from {inventory_env_path}")
# Check for local backend/.env (Legacy/Fragmented)
elif os.path.exists(backend_env_path):
load_dotenv(backend_env_path)
log.info(f"ℹ️ Loaded local configuration from {backend_env_path}")
else:
log.warning("⚠️ No .env config files found. Relying on system environment variables.")
# Auto-run if imported
load_config()

27
backend/main.py
View File

@@ -1,4 +1,5 @@
import os
from . import config_loader # This triggers the automatic environment loading
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from slowapi import Limiter
@@ -26,9 +27,9 @@ ALLOWED_ORIGINS = [o.strip() for o in _raw_origins.split(",") if o.strip()]
# Automatically add origins based on network_config.env variables if present
server_ip = os.environ.get("SERVER_IP")
front_port = os.environ.get("FRONTEND_PORT", "8907")
front_ssl_port = os.environ.get("FRONTEND_SSL_PORT", "8909")
back_ssl_port = os.environ.get("BACKEND_SSL_PORT", "8908")
front_port = os.environ.get("FRONTEND_PORT", "8917")
front_ssl_port = os.environ.get("FRONTEND_SSL_PORT", "8919")
back_ssl_port = os.environ.get("BACKEND_SSL_PORT", "8918")
# Always allow localhost
defaults = [
@@ -121,7 +122,25 @@ def startup_event():
if not existing:
db.add(models.SystemSetting(key="ai_extraction_prompt", value=final_prompt))
db.commit()
log.info("Initialized default AI prompt.")
log.info("Initialized default AI prompt in database.")
# Refresh existing after commit
existing = db.query(models.SystemSetting).filter(models.SystemSetting.key == "ai_extraction_prompt").first()
# [NEW] Sync/Initialize AI prompt file in /config
from .database import BASE_DIR
PROJECT_ROOT = os.path.dirname(BASE_DIR)
PROMPT_FILE_PATH = os.path.join(PROJECT_ROOT, "config", "ai_prompt.md")
if not os.path.exists(PROMPT_FILE_PATH):
try:
os.makedirs(os.path.dirname(PROMPT_FILE_PATH), exist_ok=True)
# Use DB value (which we just ensured exists)
current_val = existing.value if existing else final_prompt
with open(PROMPT_FILE_PATH, 'w', encoding='utf-8') as f:
f.write(current_val)
log.info(f"✅ Initialized AI prompt configuration file: {PROMPT_FILE_PATH}")
except Exception as fe:
log.error(f"❌ Failed to initialize AI prompt file: {fe}")
defaults = {
"backup_retention_count": "10",

34
backend/routers/admin_db.py
View File

@@ -135,16 +135,30 @@ async def import_database(
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
import os
from ..database import BASE_DIR
PROJECT_ROOT = os.path.dirname(BASE_DIR)
PROMPT_FILE_PATH = os.path.join(PROJECT_ROOT, "config", "ai_prompt.md")
@router.get("/settings/prompt")
def get_ai_prompt(
db: Session = Depends(get_db),
current_admin: auth.TokenData = Depends(auth.get_current_admin)
):
"""Get the current AI extraction prompt."""
"""Get the current AI extraction prompt (prioritizes config file)."""
# 1. Try file first
if os.path.exists(PROMPT_FILE_PATH):
try:
with open(PROMPT_FILE_PATH, 'r', encoding='utf-8') as f:
return {"value": f.read().strip(), "source": "file"}
except Exception as e:
print(f"Error reading prompt file: {e}")
# 2. Fallback to DB
setting = db.query(models.SystemSetting).filter(models.SystemSetting.key == "ai_extraction_prompt").first()
if not setting:
return {"value": ""}
return {"value": setting.value}
return {"value": "", "source": "none"}
return {"value": setting.value, "source": "database"}
@router.post("/settings/prompt")
def update_ai_prompt(
@@ -152,11 +166,21 @@ def update_ai_prompt(
db: Session = Depends(get_db),
current_admin: auth.TokenData = Depends(auth.get_current_admin)
):
"""Update the AI extraction prompt."""
"""Update the AI extraction prompt (writes to both file and DB)."""
value = payload.get("value")
if value is None:
raise HTTPException(status_code=400, detail="Value required")
# 1. Update File (Primary)
try:
os.makedirs(os.path.dirname(PROMPT_FILE_PATH), exist_ok=True)
with open(PROMPT_FILE_PATH, 'w', encoding='utf-8') as f:
f.write(value)
except Exception as e:
print(f"Failed to write prompt file: {e}")
# We continue to DB update anyway
# 2. Update DB (Backup/Sync)
existing = db.query(models.SystemSetting).filter(models.SystemSetting.key == "ai_extraction_prompt").first()
if existing:
existing.value = value
@@ -164,4 +188,4 @@ def update_ai_prompt(
db.add(models.SystemSetting(key="ai_extraction_prompt", value=value))
db.commit()
return {"status": "success"}
return {"status": "success", "file_updated": os.path.exists(PROMPT_FILE_PATH)}

16
backend/routers/categories.py
View File

@@ -18,21 +18,7 @@ def get_categories(
current_user: auth.TokenData = Depends(auth.get_current_user)
):
"""[C-01] List of categories — only for authenticated users."""
categories = db.query(models.Category).all()
# Auto-seed if empty with defaults mentioned by user
if not categories:
defaults = [
{"name": "Connectors", "description": "Conectica: cables, adapters, plugs"},
{"name": "Spare Parts", "description": "Piese de schimb: specific components"},
{"name": "Tools", "description": "Hand and power tools"},
{"name": "Consumables", "description": "One-time use items"}
]
for d in defaults:
cat = models.Category(**d)
db.add(cat)
db.commit()
return db.query(models.Category).all()
return categories
return db.query(models.Category).all()
@router.post("/", response_model=schemas.Category)
def create_category(