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

Build [v1.9.19]

Browse Source
This commit is contained in:
Daniel Bedeleanu
2026-04-14 20:44:01 +03:00
parent fcb187974e
commit 00ee4cf9c5
38 changed files with 2059 additions and 157 deletions
Download Patch File Download Diff File Expand all files Collapse all files

114
backend/ai_vision.py
View File

@@ -1,6 +1,5 @@
import os
from dotenv import load_dotenv
from .ai import gemini, claude
from . import models
from .database import SessionLocal
# Load environment variables from the directory where this file resides
base_dir = os.path.dirname(os.path.abspath(__file__))
@@ -12,51 +11,74 @@ def extract_label_info(image_bytes: bytes, mode: str = "item"):
Orchestrates extraction across multiple AI providers.
Modes: 'item' (full technical extraction), 'box' (container discovery)
"""
if mode == "box":
prompt = """
Identify the CONTAINER or BOX name from this image.
Look for large, prominent, bold, or hand-written text that identifies a storage unit.
Ignore small technical details, quantities, or fine print.
db = SessionLocal()
try:
if mode == "box":
prompt = """
Identify the CONTAINER or BOX name from this image.
Look for large, prominent, bold, or hand-written text that identifies a storage unit.
Ignore small technical details, quantities, or fine print.
Return ONLY a valid JSON object:
{
"box_label": "The identified container name",
"name": "Same as box_label",
"category": "Storage",
"description": "Brief description if useful",
"quantity": 1
}
"""
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 Gemini
result = gemini.extract(image_bytes, prompt)
Return ONLY a valid JSON object:
{
"box_label": "The identified container name",
"name": "Same as box_label",
"category": "Storage",
"specs": "Brief description if useful",
"quantity": 1
}
"""
else:
prompt = """
Extract technical inventory information from this label image.
if result:
# Map user-defined prompt keys to model fields if needed
# User keys: Item, Type, Description, Category, Connector, Size, Color, PartNr, OCR
mapping = {
"Item": "name",
"Type": "type",
"Description": "description",
"Category": "category",
"Connector": "connector",
"Size": "size",
"Color": "color",
"PartNr": "part_number",
"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]
# 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
finally:
db.close()
CRITICAL INSTRUCTIONS:
1. Look at the most prominent text (usually top 1-2 rows). This is the product NAME and MODEL.
2. Extract the PART NUMBER (P/N, Model No, Type). If no explicit Part Number is found, synthesize one from the most unique identifier in the header (e.g. 'OM4-MMF-DX').
3. Separate the COLOR (e.g. Turquoise, Yellow, Black).
4. Extract CATEGORY based on the item type (e.g. Patchcord, SFP, Connector).
5. Extract technical SPECS (e.g. '2.0mm', '10G', '850nm').
Return ONLY a valid JSON object:
{
"name": "Full descriptive name from header",
"part_number": "Unique identifier for fast scanning",
"category": "Broad category",
"color": "Color if present",
"specs": "Brief tech specs list",
"barcode": "Barcode value if visible",
"quantity": 1
}
"""
# 1. Try Gemini
result = gemini.extract(image_bytes, prompt)
if result:
# Maintenance: Ensure fields are mapped if mode was box
if mode == "box" and "box_label" in result and "name" not in result:
result["name"] = result["box_label"]
return result
# 2. Try Claude (Fallback) - Note: Mapping logic would need to be replicated here if enabled
# For now, keeping it simple
return {"error": "AI extraction failed or no data returned. Check your API key and Prompt."}
# 2. Try Claude (Fallback)
result = claude.extract(image_bytes, prompt)

33
backend/db_manager.py
View File

@@ -136,3 +136,36 @@ class DbManager:
except Exception as e:
logger.error(f"[DB] Restore failed: {str(e)}")
raise Exception(f"Restore failed: {str(e)}")
@staticmethod
def export_db() -> str:
"""Returns the path to the active database file for download."""
active_db_path = os.path.join(DATA_DIR, "inventory.db")
if not os.path.exists(active_db_path):
raise Exception("Database file not found")
return active_db_path
@staticmethod
def import_db(file_bytes: bytes, db: Session, user_id: int) -> bool:
"""
Overwrites the active database with the provided file bytes.
Creates a safety rollback backup first.
"""
active_db_path = os.path.join(DATA_DIR, "inventory.db")
try:
# 1. Safety backup
DbManager.create_backup(db, label="import_rollback", user_id=user_id)
# 2. Close pool connections
engine.dispose()
# 3. Write new file
with open(active_db_path, "wb") as f:
f.write(file_bytes)
logger.warning(f"[DB] IMPORT COMPLETED by user_id={user_id}")
return True
except Exception as e:
logger.error(f"[DB] Import failed: {str(e)}")
raise Exception(f"Import failed: {str(e)}")

48
backend/main.py
View File

@@ -65,7 +65,9 @@ if extra_origins_raw:
if combo not in ALLOWED_ORIGINS:
ALLOWED_ORIGINS.append(combo)
log.info(f"CORS allowed origins: {ALLOWED_ORIGINS}")
log.info("🔒 [SECURITY] CORS configuration initialized.")
for origin in ALLOWED_ORIGINS:
log.info(f" -> Allowed: {origin}")
# Add CORS middleware FIRST (before rate limiter)
app.add_middleware(
@@ -92,6 +94,50 @@ def startup_event():
scheduler.start()
sync_scheduler_config()
# [NEW] Initialize default system settings
from .database import SessionLocal
db = SessionLocal()
try:
# Default AI Prompt from User Request
default_prompt = (
"identify and summarise the minimal necessary information for a quick description if item. "
"I need the following output - <field name> : the result from you.\n"
"For any field, do not add comments in parenthesis. \n\n"
"Item: in three words type of this item\n"
"Type: what type of item is, like \"spare parts\", \"consumables\", \"patch cords\" etc.\n"
"Description: description (max 5 words)\n"
"Category: category, if any\n"
"Connector: connectors\n"
"Size: size or length\n"
"Color: color if useful\n"
"PartNr: part number if any\n"
"OCR: identification string for local OCR matching"
)
# Wrap in JSON instructions for reliable parsing
final_prompt = f"IMAGE ANALYSIS INSTRUCTIONS:\n{default_prompt}\n\nIMPORTANT: Return ONLY a valid JSON object with the keys: Item, Type, Description, Category, Connector, Size, Color, PartNr, OCR."
existing = db.query(models.SystemSetting).filter(models.SystemSetting.key == "ai_extraction_prompt").first()
if not existing:
db.add(models.SystemSetting(key="ai_extraction_prompt", value=final_prompt))
db.commit()
log.info("Initialized default AI prompt.")
defaults = {
"backup_retention_count": "10",
"backup_schedule_hour": "3",
"backup_schedule_freq_days": "1"
}
for key, val in defaults.items():
if not db.query(models.SystemSetting).filter(models.SystemSetting.key == key).first():
db.add(models.SystemSetting(key=key, value=val))
log.info(f"Initialized default setting: {key}")
db.commit()
except Exception as e:
log.error(f"Failed to initialize settings: {e}")
finally:
db.close()
@app.get("/")
def read_root():
return {"message": "Inventory API is running"}

10
backend/models.py
View File

@@ -23,6 +23,12 @@ class Category(Base):
items = relationship("Item", back_populates="category_rel")
class Color(Base):
__tablename__ = "colors"
id = Column(Integer, primary_key=True, index=True)
name = Column(String, unique=True, index=True)
class Item(Base):
__tablename__ = "items"
@@ -36,6 +42,10 @@ class Item(Base):
category_rel = relationship("Category", back_populates="items")
part_number = Column(String, index=True, nullable=True)
color = Column(String, index=True, nullable=True)
description = Column(String, nullable=True)
connector = Column(String, nullable=True)
size = Column(String, nullable=True)
ocr_text = Column(Text, nullable=True)
specs = Column(Text, nullable=True)
quantity = Column(Float, default=0.0)
min_quantity = Column(Float, default=1.0)

62
backend/routers/admin_db.py
View File

@@ -5,6 +5,8 @@ from .. import models, schemas, auth
from ..database import get_db
from ..db_manager import DbManager
from ..scheduler import sync_scheduler_config
from fastapi.responses import FileResponse
from fastapi import UploadFile, File
router = APIRouter(
prefix="/admin/db",
@@ -103,3 +105,63 @@ def update_db_settings(
# Re-trigger scheduler sync
sync_scheduler_config()
return settings
@router.get("/export")
def export_database(
current_admin: auth.TokenData = Depends(auth.get_current_admin)
):
"""Download the current database file."""
try:
path = DbManager.export_db()
return FileResponse(
path,
media_type="application/x-sqlite3",
filename="inventory_export.db"
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.post("/import")
async def import_database(
file: UploadFile = File(...),
db: Session = Depends(get_db),
current_admin: auth.TokenData = Depends(auth.get_current_admin)
):
"""Upload and replace the current database. DANGEROUS."""
contents = await file.read()
try:
DbManager.import_db(contents, db, user_id=current_admin.sub)
return {"status": "success", "message": "Database successfully imported and replaced."}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@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."""
setting = db.query(models.SystemSetting).filter(models.SystemSetting.key == "ai_extraction_prompt").first()
if not setting:
return {"value": ""}
return {"value": setting.value}
@router.post("/settings/prompt")
def update_ai_prompt(
payload: dict,
db: Session = Depends(get_db),
current_admin: auth.TokenData = Depends(auth.get_current_admin)
):
"""Update the AI extraction prompt."""
value = payload.get("value")
if value is None:
raise HTTPException(status_code=400, detail="Value required")
existing = db.query(models.SystemSetting).filter(models.SystemSetting.key == "ai_extraction_prompt").first()
if existing:
existing.value = value
else:
db.add(models.SystemSetting(key="ai_extraction_prompt", value=value))
db.commit()
return {"status": "success"}

29
backend/routers/items.py
View File

@@ -97,10 +97,18 @@ def create_item(
current_user: auth.TokenData = Depends(auth.get_current_user)
):
"""[C-01] Create item — only for authenticated users. [M-02] user_id from token."""
# Check if barcode exists
db_item = db.query(models.Item).filter(models.Item.barcode == item.barcode).first()
if db_item:
raise HTTPException(status_code=400, detail="Barcode already registered")
# [AUTO-PERSIST] Create Category/Color if not exists
if item.category:
cat = db.query(models.Category).filter(models.Category.name == item.category).first()
if not cat:
db.add(models.Category(name=item.category))
db.commit()
if item.color:
col = db.query(models.Color).filter(models.Color.name == item.color).first()
if not col:
db.add(models.Color(name=item.color))
db.commit()
db_item = models.Item(**item.model_dump())
db.add(db_item)
@@ -148,6 +156,19 @@ def update_item(
if not db_item:
raise HTTPException(status_code=404, detail="Item not found")
# [AUTO-PERSIST] Create Category/Color if not exists
if item.category:
cat = db.query(models.Category).filter(models.Category.name == item.category).first()
if not cat:
db.add(models.Category(name=item.category))
db.commit()
if item.color:
col = db.query(models.Color).filter(models.Color.name == item.color).first()
if not col:
db.add(models.Color(name=item.color))
db.commit()
update_data = item.model_dump(exclude_unset=True)
for key, value in update_data.items():
setattr(db_item, key, value)

17
backend/schemas.py
View File

@@ -51,6 +51,19 @@ class Category(CategoryBase):
class Config:
from_attributes = True
# --- Colors ---
class ColorBase(BaseModel):
name: str
class ColorCreate(ColorBase):
pass
class Color(ColorBase):
id: int
class Config:
from_attributes = True
# --- Items ---
class ItemBase(BaseModel):
name: str
@@ -60,6 +73,10 @@ class ItemBase(BaseModel):
barcode: str
part_number: Optional[str] = None
color: Optional[str] = None
description: Optional[str] = None
connector: Optional[str] = None
size: Optional[str] = None
ocr_text: Optional[str] = None
specs: Optional[str] = None
quantity: float = 0.0
min_quantity: float = 1.0

51
backend/scripts/init_settings.py Normal file
View File

@@ -0,0 +1,51 @@
import sys
import os
# Add parent directory to path
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from database import SessionLocal, engine
import models
def init_settings():
db = SessionLocal()
try:
# Default AI Prompt from User Request
default_prompt = """identify and summarise the minimal necessary information for a quick description if item. I need the following output - <field name> : the result from you.
For any field, do not add comments in parenthesis.
Item: in three words type of this item
Type: what type of item is, like "spare parts", "consumables", "patch cords" etc.
Description: description (max 5 words)
Category: category, if any
Connector: connectors
Size: size or length
Color: color if useful
PartNr: part number if any
OCR: identification string for local OCR matching"""
# We will wrap this in a instruction to return JSON for easier parsing while keeping the user's content
final_prompt = f"IMAGE ANALYSIS INSTRUCTIONS:\n{default_prompt}\n\nIMPORTANT: Return ONLY a valid JSON object with the keys: Item, Type, Description, Category, Connector, Size, Color, PartNr, OCR."
existing = db.query(models.SystemSetting).filter(models.SystemSetting.key == "ai_extraction_prompt").first()
if not existing:
db.add(models.SystemSetting(key="ai_extraction_prompt", value=final_prompt))
print("Added default AI prompt setting.")
# Add default backup settings if missing
defaults = {
"backup_retention_count": "10",
"backup_schedule_hour": "3",
"backup_schedule_freq_days": "1"
}
for key, val in defaults.items():
if not db.query(models.SystemSetting).filter(models.SystemSetting.key == key).first():
db.add(models.SystemSetting(key=key, value=val))
print(f"Added default setting: {key}")
db.commit()
finally:
db.close()
if __name__ == "__main__":
init_settings()

49
backend/scripts/migrate_v4_v5.py Normal file
View File

@@ -0,0 +1,49 @@
import sqlite3
import os
from ..database import db_path
from ..logger import log
def migrate():
"""
Migration script to upgrade the items table from schema v4 to v5.
Adds columns: description, connector, size, ocr_text.
"""
log.info(f"🚀 Starting database migration on {db_path}...")
if not os.path.exists(db_path):
log.error(f"❌ Database file not found at {db_path}")
return
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
try:
# Check existing columns
cursor.execute("PRAGMA table_info(items)")
columns = [row[1] for row in cursor.fetchall()]
new_columns = [
("description", "VARCHAR"),
("connector", "VARCHAR"),
("size", "VARCHAR"),
("ocr_text", "TEXT")
]
for col_name, col_type in new_columns:
if col_name not in columns:
log.info(f"➕ Adding column '{col_name}' to 'items' table...")
cursor.execute(f"ALTER TABLE items ADD COLUMN {col_name} {col_type}")
else:
log.info(f"✔️ Column '{col_name}' already exists.")
conn.commit()
log.info("✅ Migration completed successfully.")
except Exception as e:
log.error(f"❌ Migration failed: {e}")
conn.rollback()
finally:
conn.close()
if __name__ == "__main__":
migrate()