sapl/sapl
2
0
Fork 0
Code Issues 182 Pull Requests 6 Actions Packages Projects Releases Wiki Activity
Files
177f48030020661079c810a90ecce5abfeb968f2
sapl/backend/main.py
T
Ajay SableandClaude Opus 4.8 9f03ce3874 feat(tna): P1 authoring flow — Create TNA from a no-TNA order → edit → approve → push 
Adds the merchant authoring flow on top of the existing recommender:
- POST /tna/draft: propose a TNA for a no-TNA order, with optional template
  override + fabric-lead-time override (recommender extended for both).
- PUT /tna/recommend/{key}/schedule: persist planner date/owner edits (frozen
  once accepted); sets edited flag.
- POST /tna/push/{key}: push an approved plan's template to Stitch, gated on
  Template-Readiness — returns blocked + missing tasks when the tenant isn't
  seeded (no silent failure). Distinct path avoids the /{decision} route clash.
- TnaRecommendation gains pcd_date/edited/stitch_template_id/push_status/
  push_detail (idempotent migration + boot self-heal in services/tna_authoring).
- Frontend: TnaCreate page (proposal + template picker + fabric-lead-time input +
  editable milestone table + approve→push with readiness gate), wired from the
  "Create TNA" action on the no-TNA cards.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-09 16:01:06 +05:30

619 lines
28 KiB
Python
Raw Blame History

import sys
import asyncio
import logging
from contextlib import asynccontextmanager
from pathlib import Path
# On Windows, SelectorEventLoop (uvicorn default) cannot launch subprocesses.
# Playwright/Camoufox needs subprocess support to launch browsers.
# ProactorEventLoop is the Windows default since Python 3.8 and is fully compatible with uvicorn.
if sys.platform == "win32":
asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())
from fastapi import FastAPI, Depends
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from config import settings
from database import Base, engine
from routers import garments, search, review, ingest, moodboard, scraper, tna, auth, admin, collections, plm, mail, users, concept, product_studio, fabrics, hermes
from routers.auth import require_permission
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Service singletons — initialised once at startup, injected into routers
# ---------------------------------------------------------------------------
from models.trend import TrendSearch, ScrapedItem # noqa: F401 — registers tables
from models.tna_stitch import StitchOrder, StitchMilestone # noqa: F401 — registers tables
from models.tna_template import TnaTemplate, TnaRecommendation # noqa: F401 — registers TNA recommendation tables
from models.tna_approval import TnaApprovalEvent # noqa: F401 — registers tna_approval_events table
from models.user import User # noqa: F401 — registers users table
from models.collection import Collection, CollectionItem # noqa: F401 — registers boards tables
from models.moodboard import MoodboardDecode # noqa: F401 — registers moodboard_decodes table
from models.fabric import Fabric # noqa: F401 — registers fabrics table
from services.embedding_service import EmbeddingService
from services.tagging_service import TaggingService
from services.similarity_service import SimilarityService
_embedding_svc: EmbeddingService | None = None
_tagging_svc: TaggingService | None = None
_similarity_svc: SimilarityService | None = None
# Taxonomy — static data served for frontend dropdowns
TAXONOMY = {
"category": [
"Shirt", "Blouse", "T-Shirt", "Polo", "Sweatshirt", "Hoodie", "Overshirt",
"Top", "Vest", "Trousers", "Jeans", "Joggers", "Shorts", "Skirt",
"Outerwear", "Blazer", "Dress", "Jumpsuit", "Co-ord Set",
"Kurta", "Swimwear", "Nightwear", "Loungewear",
],
"group": ["Menswear", "Womenswear", "Boyswear", "Girlswear", "Unisex"],
"macro_family": ["Topwear", "Bottomwear", "Onepiece", "Outerwear"],
"fabric_family": ["Woven", "Knit", "Denim", "Non-woven"],
"silhouette": ["Slim", "Regular", "Relaxed", "Oversized", "Cropped", "Longline"],
"coloration_method": [
"Solid dyed", "Yarn-dyed", "Printed — Rotary", "Printed — Digital",
"Printed — Discharge", "Printed — Pigment", "Printed — Placement",
"AOP", "Ombré", "Tie-dye", "Raw / Natural",
],
"neckline": [
"Point collar", "Button-down collar", "Spread collar", "Band collar",
"Camp collar", "Half-placket", "V-neck", "Round neck", "Boat neck",
"Tie-neck", "Ruffle collar", "Peter Pan collar", "Mandarin collar",
"Square neck", "Halter", "Off-shoulder", "One-shoulder", "Cowl neck",
"Crew neck", "Henley", "Polo collar", "Mock neck", "Scoop neck", "none",
],
"weave_knit_structure": [
"Plain weave", "Twill", "Satin", "Dobby", "Jacquard", "Oxford",
"Chambray", "Seersucker", "Leno", "Fil-à-fil", "Herringbone", "Corduroy",
"Flannel", "Voile", "Crepe", "Canvas", "Ripstop",
"Single jersey", "Interlock", "French terry", "Fleece", "Rib", "Piqué",
"Waffle", "Ponte", "Tricot", "Raschel", "Jacquard knit", "Terry", "Velour",
"Rigid denim", "Stretch denim", "Chambray denim", "Selvedge", "Bull denim",
],
"sleeve": [
"Long sleeve", "Short sleeve", "Three-quarter sleeve", "Rolled sleeve",
"Sleeveless", "Cap sleeve", "Flutter sleeve", "Bell sleeve", "Bishop sleeve",
"Puff sleeve", "Raglan sleeve", "Drop shoulder", "Dolman sleeve",
],
"value_addition": [
"Embroidery — All-over", "Embroidery — Placement", "Embroidery — Appliqué",
"Embroidery — Schiffli", "Embroidery — Bead/Sequin",
"Print — Graphic", "Print — Stripe", "Print — Floral", "Print — Abstract",
"Print — Tropical", "Print — Check/Plaid",
"Wash — Enzyme", "Wash — Stone", "Wash — Acid", "Wash — Bleach",
"Wash — Silicone", "Wash — Ozone",
"Overdye", "Laser treatment", "Smocking", "Pleating", "Quilting",
"Crochet", "Lace", "Fringe", "Patchwork", "none",
],
"construction_details": [
"Chest pocket — Patch", "Chest pocket — Welt", "Chest pocket — Flap", "No pocket",
"Button-through", "Half-placket", "Zip-through", "Quarter-zip",
"Drawstring waist", "Elasticated waist", "Belt loops", "Cargo pockets",
"Lined", "Unlined", "Side slit", "Back yoke", "Epaulettes", "Hood",
"Ribbed hem", "Wrap front", "Tiered", "Ruching", "Pintuck",
],
"length": [
"Mini", "Midi", "Maxi", "Knee-length", "Full length",
"Cropped", "Bermuda", "Short", "Micro short",
],
"closure": [
"button_down_front", "button_down_back", "pullover", "wrap_crossover",
"tie_front", "tie_back", "zip_back", "zip_side", "lace_up_front", "none",
],
"hem": [
"straight", "curved_high_low", "asymmetric", "elasticated",
"tiered_ruffle", "raw_frayed", "scallop",
],
"fabric_texture": [
"woven_smooth", "crinkled_crushed", "hammered_satin", "jacquard_textured",
"sheer_semi_sheer", "knit_jersey", "denim", "linen_look", "lace", "broderie",
],
"design_dna": [
"romantic_feminine", "heritage_check_plaid",
"animal_elevated", "classic_minimal", "mixed",
],
"season": ["AW26", "SS26", "AW25", "SS25"],
"design_team_city": ["mumbai", "bangalore", "spain", "delhi", "chennai", "kolkata", "unknown"],
"status": ["prototype", "approved", "in_production", "archived"],
}
@asynccontextmanager
async def lifespan(app: FastAPI):
global _embedding_svc, _tagging_svc, _similarity_svc
# Security guard — never let prod run on the forgeable built-in default secret.
if settings.secret_key == "sapl-garment-intel-dev-secret-change-in-prod-2026":
logger.critical(
"SECRET_KEY is the built-in DEV default — auth tokens are FORGEABLE. "
"Set a strong SECRET_KEY in the environment before any real use."
)
# Create all DB tables on startup
Base.metadata.create_all(bind=engine)
# Self-heal RBAC columns on a pre-existing DB (create_all can't ALTER).
from services.rbac import ensure_rbac_schema
ensure_rbac_schema(engine)
from services.fabric_seed import ensure_fabric_schema
ensure_fabric_schema(engine)
from services.tna_authoring import ensure_tna_authoring_schema
ensure_tna_authoring_schema(engine)
logger.info("Database tables ready")
# Seed dummy users if none exist
from database import SessionLocal as _SL
from models.user import User as _User
from services.auth_service import hash_password as _hash
_db = _SL()
try:
if _db.query(_User).count() == 0:
seed = [
("pratik@sonalapparels.in", "atelier2026", "Pratik", "admin"),
("suresh@sonalapparels.in", "planning2026", "Suresh", "planner"),
("ajay@sonalapparels.in", "dev2026", "Ajay Sable", "admin"),
("nilesh@sonalapparels.in", "audit2026", "Nilesh", "auditor"),
]
for email, pw, name, role in seed:
_db.add(_User(email=email, name=name, role=role, password_hash=_hash(pw)))
_db.commit()
logger.info(f"Seeded {len(seed)} users")
# RBAC: ensure default departments + roles exist and the seeded users are
# mapped onto them (idempotent — safe on every boot).
from services.rbac import seed_rbac
seed_rbac(_db)
# Seed the Fabric Library from the committed snapshot (idempotent —
# only fills an empty fabrics table; restores missing images otherwise).
from services.fabric_seed import seed_fabrics, backfill_source_hashes
seed_fabrics(_db)
backfill_source_hashes(_db)
finally:
_db.close()
# Ensure image directory exists
Path(settings.images_path).mkdir(parents=True, exist_ok=True)
# Initialise services once — avoids ChromaDB thread-safety race on first request
logger.info("Initialising services...")
_embedding_svc = EmbeddingService()
_tagging_svc = TaggingService()
_similarity_svc = SimilarityService(_embedding_svc)
# Share instances with routers via module-level injection
garments.set_services(_tagging_svc, _embedding_svc, _similarity_svc)
search.set_embedding_service(_embedding_svc)
review.set_embedding_service(_embedding_svc)
ingest.set_services(_tagging_svc, _embedding_svc, _similarity_svc)
moodboard.set_services(_tagging_svc, _embedding_svc)
logger.info("SAPL Garment Intelligence API started")
import asyncio
# All integration syncs run in the BACKGROUND so the API starts serving at
# once. Previously the lifespan AWAITED the initial Stitch pull (50+ pages of
# style-plan-master) before yielding — uvicorn won't accept requests until the
# lifespan reaches `yield`, so the whole frontend sat empty until it finished.
# Now startup yields immediately and the data fills in as each sync lands.
# Stitch TNA: initial sync, then the 2-min loop.
if settings.stitch_username and settings.stitch_password:
async def _stitch_bootstrap():
from services.stitch_service import stitch_sync_loop, sync_from_stitch
logger.info("Running initial Stitch TNA sync (background)...")
try:
await sync_from_stitch()
except Exception as e:
logger.warning(f"Initial Stitch sync failed (non-fatal): {e}")
await stitch_sync_loop()
asyncio.create_task(_stitch_bootstrap())
else:
logger.warning("Stitch credentials not set — TNA sync disabled.")
# TNA template master (the recommendation engine's 17 templates), best-effort.
if settings.stitch_api_key:
async def _template_bootstrap():
from services.tna_recommend import sync_templates
from database import SessionLocal as _TSL
try:
_tdb = _TSL()
await sync_templates(_tdb)
_tdb.close()
except Exception as e:
logger.warning(f"Initial TNA template sync failed (non-fatal): {e}")
asyncio.create_task(_template_bootstrap())
# WFX order-mirror: the sync is synchronous (urllib), so run it in a worker
# thread — that keeps it off the event loop entirely (it would otherwise block
# every request for the call's duration). First sync fires at once, then 30s.
if settings.wfx_token_id:
from services.wfx_service import sync_wfx_orders
from database import SessionLocal as _WSL
def _wfx_sync_once():
d = _WSL()
try:
sync_wfx_orders(d)
finally:
d.close()
async def _wfx_sync_loop():
while True:
try:
await asyncio.to_thread(_wfx_sync_once)
except Exception as e:
logger.warning(f"WFX mirror sync failed: {e}")
await asyncio.sleep(30) # 30s — a new WFX lot shows up near-instantly
asyncio.create_task(_wfx_sync_loop())
logger.info("WFX mirror sync loop started (background).")
else:
logger.warning("WFX token not set — WFX mirror disabled.")
# Start Gmail order-mail ingest (Project P1: Gmail → WFX). Reads the BOM label,
# AI-extracts lot + dates. Runs in a thread so blocking IMAP/Gemini calls never
# stall the event loop. Extraction only fires on NEW mails (deduped), so steady
# state is a cheap IMAP poll.
if settings.mail_auth_user and settings.mail_app_password:
from services.mail_service import sync_mail, idle_listen
from database import SessionLocal as _MSL
import asyncio
import threading
def _mail_sync_once():
d = _MSL()
try:
res = sync_mail(d)
if res.get("new"):
logger.info(f"Mail ingest: {res['new']} new mail(s) fetched.")
finally:
d.close()
# Instant push via IMAP IDLE — fires sync the moment mail lands in BOM.
threading.Thread(target=lambda: idle_listen(_mail_sync_once), daemon=True).start()
logger.info("Mail IDLE listener thread started (instant ingest).")
async def _mail_sync_loop():
# Small startup delay so the API is serving before the first poll.
await asyncio.sleep(10)
while True:
try:
def _run():
d = _MSL()
try:
return sync_mail(d)
finally:
d.close()
res = await asyncio.to_thread(_run)
if res.get("new"):
logger.info(f"Mail ingest: {res['new']} new mail(s) fetched.")
except Exception as e:
logger.warning(f"Mail ingest sync failed: {e}")
await asyncio.sleep(2 * 60) # 2 minutes
asyncio.create_task(_mail_sync_loop())
logger.info("Mail ingest sync loop started.")
else:
logger.warning("Mail credentials not set — Gmail ingest disabled.")
yield
logger.info("Shutting down")
app = FastAPI(
title="SAPL Garment Intelligence API",
description="AI-powered garment tagging and semantic search for SAPL AW26",
version="1.0.0",
lifespan=lifespan,
)
# Local dev origins + any extra origins from settings.cors_origins (prod frontend)
_cors_origins = ["http://localhost:5173", "http://localhost:5174", "http://localhost:3000"]
_cors_origins += [o.strip() for o in settings.cors_origins.split(",") if o.strip()]
app.add_middleware(
CORSMiddleware,
allow_origins=_cors_origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Static file serving — ensure the dirs exist before mounting. StaticFiles is
# constructed at import time (before the lifespan startup hook runs), so a fresh
# checkout without a data/ dir would otherwise crash on mount.
from pathlib import Path as _Path
_Path(settings.images_path).mkdir(parents=True, exist_ok=True)
app.mount("/images", StaticFiles(directory=settings.images_path), name="images")
_scraped_dir = _Path(settings.images_path).parent / "scraped"
_scraped_dir.mkdir(parents=True, exist_ok=True)
app.mount("/scraped", StaticFiles(directory=str(_scraped_dir)), name="scraped")
_concepts_dir = _Path(settings.images_path).parent / "concepts"
_concepts_dir.mkdir(parents=True, exist_ok=True)
app.mount("/concepts", StaticFiles(directory=str(_concepts_dir)), name="concepts")
# Routers
app.include_router(garments.router, prefix="/api/garments", tags=["garments"])
app.include_router(search.router, prefix="/api/search", tags=["search"])
app.include_router(review.router, prefix="/api/review", tags=["review"])
app.include_router(ingest.router, prefix="/api/ingest", tags=["ingest"])
app.include_router(moodboard.router, prefix="/api/moodboard", tags=["moodboard"])
if settings.enable_scraper:
app.include_router(scraper.router, prefix="/api/scraper", tags=["scraper"])
app.include_router(tna.router, prefix="/api/tna", tags=["tna"])
app.include_router(auth.router, prefix="/api/auth", tags=["auth"])
app.include_router(admin.router, prefix="/api/admin", tags=["admin"])
app.include_router(users.router, prefix="/api/admin", tags=["user-management"])
app.include_router(collections.router, prefix="/api/collections", tags=["collections"])
app.include_router(plm.router)
app.include_router(mail.router)
app.include_router(concept.router, prefix="/api/concept", tags=["concept"])
app.include_router(product_studio.router, prefix="/api/product-studio", tags=["product-studio"])
app.include_router(fabrics.router, prefix="/api/fabrics", tags=["fabrics"])
app.include_router(hermes.router, prefix="/api/hermes", tags=["hermes"])
# prefix /api/plm declared on the router
@app.get("/api/health", tags=["meta"])
def health():
return {"status": "ok", "version": "2.0.0", "collection": "SAPL AW26 Womenswear"}
@app.get("/api/taxonomy", tags=["meta"])
def get_taxonomy():
return TAXONOMY
@app.get("/api/stats", tags=["meta"])
def get_stats():
from sqlalchemy import func, text
from database import SessionLocal
from models.garment import Garment, GarmentTag
from models.review_queue import ReviewQueueItem
db = SessionLocal()
try:
total = db.query(Garment).filter(Garment.status != "archived").count()
# By category
type_rows = (
db.query(GarmentTag.field_value, func.count(GarmentTag.id))
.filter(GarmentTag.field_name == "category")
.group_by(GarmentTag.field_value)
.all()
)
by_type = {row[0]: row[1] for row in type_rows}
# By design_dna
dna_rows = (
db.query(GarmentTag.field_value, func.count(GarmentTag.id))
.filter(GarmentTag.field_name == "design_dna")
.group_by(GarmentTag.field_value)
.all()
)
by_dna = {row[0]: row[1] for row in dna_rows}
# By series
series_rows = (
db.query(Garment.series_code, func.count(Garment.id))
.group_by(Garment.series_code)
.all()
)
by_series = {row[0]: row[1] for row in series_rows}
pending_review = db.query(ReviewQueueItem).filter(ReviewQueueItem.status == "pending").count()
ai_auto = db.query(Garment).filter(Garment.review_status == "ai_tagged_auto").count()
human_verified = db.query(Garment).filter(Garment.review_status == "human_verified").count()
return {
"total_garments": total,
"by_category": by_type,
"by_design_dna": by_dna,
"by_series": by_series,
"pending_review_items": pending_review,
"ai_auto_tagged": ai_auto,
"human_verified": human_verified,
}
finally:
db.close()
@app.get("/api/analytics", tags=["meta"], dependencies=[Depends(require_permission("analytics", "view"))])
def get_analytics(
season: str | None = None,
segment: str | None = None,
status: str | None = None,
category: str | None = None, # mens | womens | kids (Garment.category)
collection: str | None = None,
fabric_family: str | None = None,
design_team_city: str | None = None,
design_dna: str | None = None, # tag-based filter
):
import json as _json
from collections import Counter
from sqlalchemy import func
from database import SessionLocal
from models.garment import Garment, GarmentTag
from models.review_queue import ReviewQueueItem
db = SessionLocal()
try:
# ── Filter options (over the whole non-archived library, so dropdowns stay stable) ──
def col_options(attr: str) -> list:
rows = (
db.query(getattr(Garment, attr), func.count(Garment.id))
.filter(Garment.status != "archived")
.group_by(getattr(Garment, attr))
.order_by(func.count(Garment.id).desc())
.all()
)
return [{"value": r[0], "count": r[1]} for r in rows if r[0]]
dna_opt_rows = (
db.query(GarmentTag.field_value, func.count(func.distinct(GarmentTag.garment_id)))
.filter(GarmentTag.field_name == "design_dna")
.group_by(GarmentTag.field_value)
.order_by(func.count(func.distinct(GarmentTag.garment_id)).desc())
.all()
)
filter_options = {
"season": col_options("season"),
"segment": col_options("segment"),
"status": col_options("status"),
"category": col_options("category"),
"collection": col_options("collection_name"),
"fabric_family": col_options("fabric_family"),
"design_team_city": col_options("design_team_city"),
"design_dna": [{"value": r[0], "count": r[1]} for r in dna_opt_rows if r[0]],
}
applied = {
k: v for k, v in {
"season": season, "segment": segment, "status": status,
"category": category, "collection": collection,
"fabric_family": fabric_family, "design_team_city": design_team_city,
"design_dna": design_dna,
}.items() if v
}
# ── Filtered garment set — every distribution below is computed over THIS set ──
gq = db.query(Garment)
gq = gq.filter(Garment.status == status) if status else gq.filter(Garment.status != "archived")
if season: gq = gq.filter(Garment.season == season)
if segment: gq = gq.filter(Garment.segment == segment)
if category: gq = gq.filter(Garment.category == category)
if collection: gq = gq.filter(Garment.collection_name == collection)
if fabric_family: gq = gq.filter(Garment.fabric_family == fabric_family)
if design_team_city: gq = gq.filter(Garment.design_team_city == design_team_city)
if design_dna:
dna_ids = [r[0] for r in db.query(GarmentTag.garment_id).filter(
GarmentTag.field_name == "design_dna", GarmentTag.field_value == design_dna).all()]
gq = gq.filter(Garment.id.in_(dna_ids))
garments = gq.all()
gids = [g.id for g in garments]
total = len(garments)
def tag_distribution(field: str) -> dict:
if not gids:
return {}
rows = (
db.query(GarmentTag.field_value, func.count(GarmentTag.id))
.filter(GarmentTag.field_name == field, GarmentTag.garment_id.in_(gids))
.group_by(GarmentTag.field_value)
.order_by(func.count(GarmentTag.id).desc())
.all()
)
return {r[0]: r[1] for r in rows}
def col_distribution(attr: str, default: str = "Unknown") -> dict:
c = Counter(getattr(g, attr) or default for g in garments)
return dict(c.most_common())
# Colour distribution — field_value is a JSON list
colour_counts: dict = {}
if gids:
colour_rows = (
db.query(GarmentTag.field_value)
.filter(GarmentTag.field_name == "colour", GarmentTag.garment_id.in_(gids))
.all()
)
for (val,) in colour_rows:
try:
for c in (_json.loads(val) if val else []):
name = c.get("colour_name", "Unknown")
key = name
if key not in colour_counts:
colour_counts[key] = {"count": 0, "hex": c.get("hex", "#888888"), "hero_count": 0}
colour_counts[key]["count"] += 1
if c.get("role", "Hero") == "Hero":
colour_counts[key]["hero_count"] += 1
except Exception:
pass
colour_list = sorted(
[{"name": k, **v} for k, v in colour_counts.items()],
key=lambda x: x["hero_count"], reverse=True,
)
# Confidence by field — avg (over the filtered set)
confidence_by_field = []
if gids:
conf_rows = (
db.query(GarmentTag.field_name, func.avg(GarmentTag.confidence), func.count(GarmentTag.id))
.filter(GarmentTag.garment_id.in_(gids))
.group_by(GarmentTag.field_name)
.order_by(func.avg(GarmentTag.confidence).asc())
.all()
)
confidence_by_field = [
{"field": r[0], "avg_confidence": round(float(r[1]), 3), "tag_count": r[2]}
for r in conf_rows
]
# Library growth over time — garments added per day + cumulative (from created_at)
day_added = Counter(
g.created_at.date().isoformat() for g in garments if g.created_at is not None
)
growth, _cum = [], 0
for d in sorted(day_added):
_cum += day_added[d]
growth.append({"date": d, "added": day_added[d], "cumulative": _cum})
# Review health — scoped to the filtered set
tq = db.query(GarmentTag).filter(GarmentTag.garment_id.in_(gids)) if gids else db.query(GarmentTag).filter(False)
total_tags = tq.count()
auto_saved = tq.filter(GarmentTag.confidence >= 0.85).count()
needs_review_tags = tq.filter(GarmentTag.confidence >= 0.65, GarmentTag.confidence < 0.85).count()
low_conf = tq.filter(GarmentTag.confidence < 0.65).count()
rq = db.query(ReviewQueueItem).filter(ReviewQueueItem.garment_id.in_(gids)) if gids else db.query(ReviewQueueItem).filter(False)
pending_q = rq.filter(ReviewQueueItem.status == "pending").count()
approved_q = rq.filter(ReviewQueueItem.status == "approved").count()
corrected_q = rq.filter(ReviewQueueItem.status == "corrected").count()
return {
"total_garments": total,
"applied_filters": applied,
"filter_options": filter_options,
"growth": growth,
# tag-based distributions
"by_design_dna": tag_distribution("design_dna"),
"by_category": tag_distribution("category"),
"by_sub_type": tag_distribution("sub_type"),
"by_coloration_method": tag_distribution("coloration_method"),
"by_value_addition": tag_distribution("value_addition"),
"by_silhouette": tag_distribution("silhouette"),
"by_neckline": tag_distribution("neckline"),
"by_sleeve": tag_distribution("sleeve"),
"by_length": tag_distribution("length"),
"by_hem": tag_distribution("hem"),
"by_closure": tag_distribution("closure"),
"by_fabric_texture": tag_distribution("fabric_texture"),
"by_embellishment": tag_distribution("embellishment"),
"by_weave": tag_distribution("weave_knit_structure"),
"by_group": tag_distribution("group"),
# garment-column distributions
"by_fabric_family": col_distribution("fabric_family"),
"by_macro_family": col_distribution("macro_family"),
"by_design_team_city": col_distribution("design_team_city"),
"by_status": col_distribution("status"),
"by_segment": col_distribution("segment"),
"colour_palette": colour_list,
"confidence_by_field": confidence_by_field,
"review_health": {
"total_tags": total_tags,
"auto_saved": auto_saved,
"needs_review": needs_review_tags,
"low_confidence": low_conf,
"auto_save_pct": round(auto_saved / total_tags * 100, 1) if total_tags else 0,
"queue_pending": pending_q,
"queue_approved": approved_q,
"queue_corrected": corrected_q,
},
}
finally:
db.close()
Reference in New Issue View Git Blame Copy Permalink