ali_mohaghegh/ERP
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update dashboard

Browse Source
This commit is contained in:
fatemeh sane
2026-10-21 16:04:07 +03:30
parent 41f93acec5
commit 8b2970cdb5
12 changed files with 1621 additions and 450 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/routers/dashboard.py
+13 -7
View File
@@ -100,12 +100,14 @@ def _what_filters(
brand_id: uuid.UUID | None,
specs_hash: str | None,
spec_id: str | None = None,
material_family: str | None = None,
) -> dict:
"""دامنه‌ی «چه چیزی» را به پارامترهای سرویس تبدیل می‌کند.
شناسه‌ی ماده اگر UUID معتبر باشد → دامنه‌ی همان ماده (همه‌ی مشخصات آن).
اگر مقدار ارسالی UUID نباشد (یعنی شناسه‌ی مشخصه فرستاده شده) یا spec_id
صریح بیاید → دامنه به همان مشخصه محدود می‌شود (specs_hash).
• شناسه‌ی ماده اگر UUID معتبر باشد → دامنه‌ی همان ماده (همه‌ی مشخصات آن).
• اگر مقدار ارسالی UUID نباشد (یعنی شناسه‌ی مشخصه فرستاده شده) یا spec_id
صریح بیاید → دامنه به همان مشخصه محدود می‌شود (specs_hash).
• material_family → همه‌ی ردیف‌های کاتالوگِ همان خانواده (سرویس حل می‌کند).
"""
resolved_material: uuid.UUID | None = None
resolved_specs = specs_hash or spec_id
@@ -119,6 +121,7 @@ def _what_filters(
"material_id": resolved_material,
"brand_id": brand_id,
"specs_hash": resolved_specs,
"material_family": (material_family or "").strip() or None,
}
@@ -136,13 +139,14 @@ async def inventory_drilldown(
brand_id: uuid.UUID | None = Query(default=None),
specs_hash: str | None = Query(default=None),
spec_id: str | None = Query(default=None, description="دامنه‌ی یک مشخصه‌ی خاص"),
material_family: str | None = Query(default=None, description="خانواده‌ی متریال (نام پایه)"),
):
"""Drill-down matrix: کل سازمان → پروژه → انبار (per-unit aggregates)."""
return await dashboard_service.inventory_drilldown(
db,
project_ids=scope.project_ids,
project_id=scope.project_id,
**_what_filters(material_id, brand_id, specs_hash, spec_id),
**_what_filters(material_id, brand_id, specs_hash, spec_id, material_family),
)
@@ -150,12 +154,13 @@ async def inventory_drilldown(
async def dead_stock(
scope: DashboardScope = Depends(dashboard_scope),
db: AsyncSession = Depends(get_db),
idle_days: int = Query(default=60, ge=1, le=365, description="آستانه رکود (روز)"),
idle_days: int = Query(default=0, ge=0, le=365, description="آستانه رکود (روز) — ۰ یعنی همه"),
limit: int = Query(default=200, ge=1, le=1000),
material_id: str | None = Query(default=None, description="UUID ماده یا شناسه‌ی مشخصه"),
brand_id: uuid.UUID | None = Query(default=None),
specs_hash: str | None = Query(default=None),
spec_id: str | None = Query(default=None, description="دامنه‌ی یک مشخصه‌ی خاص"),
material_family: str | None = Query(default=None, description="خانواده‌ی متریال (نام پایه)"),
):
"""Dead stock: موجودی مثبت بدون هیچ تراکنشی در بازه‌ی رکود."""
return await dashboard_service.dead_stock(
@@ -164,7 +169,7 @@ async def dead_stock(
project_id=scope.project_id,
idle_days=idle_days,
limit=limit,
**_what_filters(material_id, brand_id, specs_hash, spec_id),
**_what_filters(material_id, brand_id, specs_hash, spec_id, material_family),
)
@@ -179,6 +184,7 @@ async def inventory_flow(
brand_id: uuid.UUID | None = Query(default=None),
specs_hash: str | None = Query(default=None),
spec_id: str | None = Query(default=None, description="دامنه‌ی یک مشخصه‌ی خاص"),
material_family: str | None = Query(default=None, description="خانواده‌ی متریال (نام پایه)"),
):
"""Daily flow series: ورود / خروج (با لایه‌ی ضایعات) / برگشت / مانده."""
today = datetime.now(dashboard_service.TEHRAN).date()
@@ -195,7 +201,7 @@ async def inventory_flow(
date_from=datetime.combine(date_from, datetime.min.time(), tzinfo=dashboard_service.TEHRAN),
date_to=datetime.combine(date_to, datetime.min.time(), tzinfo=dashboard_service.TEHRAN),
warehouse_id=warehouse_id,
**_what_filters(material_id, brand_id, specs_hash, spec_id),
**_what_filters(material_id, brand_id, specs_hash, spec_id, material_family),
)
backend/app/services/dashboard_service.py
+228 -48
View File
@@ -31,6 +31,14 @@ from app.models.procurement import (
Quote,
QuoteProjects,
)
from app.services.material_naming import (
material_entry_labels,
material_family_key,
spec_label_from_specs,
)
# برچسب مشخصه‌ی خالی (ردیف‌های دارای specs_hash پیش‌فرض)
DEFAULT_SPEC_LABEL = "بدون مشخصه (پیش‌فرض)"
# Iran has been on a fixed UTC+03:30 (no DST) since 2022 — «امروز» in the
# daily widgets means today at this offset.
@@ -43,6 +51,21 @@ TEHRAN = timezone(timedelta(hours=3, minutes=30))
OUTBOUND_TYPES = ("WITHDRAWAL", "DAMAGE", "WASTE", "LOSS")
WASTE_TYPES = ("DAMAGE", "WASTE", "LOSS")
# دسته‌های عملیاتی «کنترل روزانه» — تنها تراکنش‌هایی که در فید امروز دیده می‌شوند
OPS_CATEGORIES = ("outbound", "returns", "waste", "purchases")
OPS_TYPES = ("WITHDRAWAL", "RETURN", "DAMAGE", "WASTE", "LOSS", "PURCHASE")
def _ops_category(tx_type: str) -> str:
"""نوع تراکنش دفتر → دسته‌ی عملیاتی (خروج/برگشت/ضایعات/پذیرش)."""
if tx_type == "WITHDRAWAL":
return "outbound"
if tx_type == "RETURN":
return "returns"
if tx_type == "PURCHASE":
return "purchases"
return "waste" # DAMAGE / WASTE / LOSS
def _stock_filters(stmt, *, material_id=None, brand_id=None, specs_hash=None):
"""«What» filters for a statement that already joins inventory→material."""
@@ -68,6 +91,21 @@ def _qty(value) -> float:
return round(float(value), 4) if value is not None else 0.0
async def _family_material_ids(db: AsyncSession, family: str) -> list[uuid.UUID]:
"""شناسه‌ی همه‌ی ردیف‌های کاتالوگِ یک خانواده.
خانواده از نام پایه‌ی ماده مشتق می‌شود (material_naming)؛ بنابراین انتخاب
«لوله پلی‌اتیلن» همه‌ی سایزهایش را با هم فیلتر می‌کند — بدون هاردکد نام.
"""
name = (family or "").strip()
if not name:
return []
rows = (
await db.execute(select(Material.id, Material.name).where(Material.status == "active"))
).all()
return [material_id for material_id, material_name in rows if material_family_key(material_name) == name]
# ---------------------------------------------------------------------------
# «What» filter cascade — Category → Material → Brand → Specs
# ---------------------------------------------------------------------------
@@ -125,22 +163,78 @@ async def filter_options(db: AsyncSession) -> dict:
unit = units.get(m.canonical_unit_id)
brands = brands_by_material.get(m.id, {})
spec_map = specs_by_material.get(m.id, {})
labels = material_entry_labels(m.name, None)
return {
"id": str(m.id),
"name": m.name,
# خانواده vs واریانت — تفکیک عمومی از روی نام کاتالوگ
"family_id": labels["family_id"],
"family_name": labels["family_name"],
"variant_label": labels["variant_label"],
"unit": {"id": unit.id, "code": unit.code, "name_fa": unit.name_fa} if unit else None,
"brands": [
{"id": str(bid), "name": bname}
for bid, bname in sorted(brands.items(), key=lambda kv: kv[1] or "")
],
"specs": [
{"hash": h, "specs": sp}
{
"hash": h,
"specs": sp,
"label": spec_label_from_specs(sp) or DEFAULT_SPEC_LABEL,
}
for h, sp in sorted(spec_map.items())
],
}
# فهرست تخت متریال‌ها — سرآغاز آبشار سه‌فیلتری (متریال → برند → مشخصات)
return {"data": {"materials": [material_entry(m) for m in materials]}}
entries = [material_entry(m) for m in materials]
# خانواده‌ها: گروه‌بندی ردیف‌های کاتالوگ بر اساس نام پایه (بدون هاردکد نام)
family_map: dict[str, dict] = {}
for entry in entries:
family = family_map.setdefault(
entry["family_id"],
{
"id": entry["family_id"],
"name": entry["family_name"],
"unit": entry["unit"],
"brands": {},
"specs": {},
"materials": [],
},
)
if family["unit"] is None:
family["unit"] = entry["unit"]
family["materials"].append(
{
"id": entry["id"],
"name": entry["name"],
"variant_label": entry["variant_label"],
}
)
for brand in entry["brands"]:
family["brands"][brand["id"]] = brand["name"]
for spec in entry["specs"]:
family["specs"][spec["hash"]] = spec["label"]
families = [
{
"id": family["id"],
"name": family["name"],
"unit": family["unit"],
"brands": [
{"id": bid, "name": bname}
for bid, bname in sorted(family["brands"].items(), key=lambda kv: kv[1] or "")
],
"specs": [
{"hash": spec_hash, "label": label}
for spec_hash, label in sorted(family["specs"].items(), key=lambda kv: kv[1] or "")
],
"materials": family["materials"],
}
for family in sorted(family_map.values(), key=lambda item: item["name"])
]
return {"data": {"materials": entries, "families": families}}
# ---------------------------------------------------------------------------
@@ -155,6 +249,7 @@ async def inventory_drilldown(
material_id=None,
brand_id=None,
specs_hash=None,
material_family=None,
) -> dict:
"""Stock quantity as a three-level tree: Material -> Project -> Warehouse.
@@ -212,6 +307,7 @@ async def inventory_drilldown(
Material.name.label("mname"),
Brand.name.label("bname"),
stock_sq.c.specs,
stock_sq.c.specs_hash,
stock_sq.c.pid,
stock_sq.c.pname,
stock_sq.c.wid,
@@ -232,6 +328,8 @@ async def inventory_drilldown(
# فیلترهای بُعد کاتالوگ، عضویت روی محور را محدود می‌کنند
if material_id is not None:
stmt = stmt.where(Material.id == material_id)
if material_family:
stmt = stmt.where(Material.id.in_(await _family_material_ids(db, material_family)))
# برند/مشخصات، ابعاد دفتر انبارند: با فعال بودنشان فقط مواد دارای همان
# موجودی روی محور می‌مانند (رفتار قبلی دست‌نخورده)
if brand_id is not None:
@@ -247,7 +345,7 @@ async def inventory_drilldown(
materials_tree: dict[str, dict] = {}
for mid, mname, bname, specs, pid, pname, wid, wname, ucode, uname_fa, qty, row_count, canon_ucode, canon_uname_fa in rows:
for mid, mname, bname, specs, specs_hash, pid, pname, wid, wname, ucode, uname_fa, qty, row_count, canon_ucode, canon_uname_fa in rows:
q = _qty(qty)
# افسانه‌ی واحدها و جمع کل؛ مواد بدون ردیف موجودی با واحد کاتالوگی و صفر
@@ -269,9 +367,25 @@ async def inventory_drilldown(
# Use full_name + unit as unique key for material level, or just full_name
mat_key = f"{mid}_{full_name}"
# برچسب‌های محور X: خانواده (نام پایه) و مشخصه/واریانت — مشتق از نام کاتالوگ
labels = material_entry_labels(mname, specs)
mat = materials_tree.setdefault(
mat_key,
{"id": mat_key, "name": full_name, "qty_by_unit": {}, "rows": 0, "_projects": {}},
{
"id": mat_key,
"name": full_name,
"material_id": str(mid),
"specs_hash": specs_hash or "",
"family_id": labels["family_id"],
"family_name": labels["family_name"],
"variant_label": labels["variant_label"],
"spec_label": labels["spec_label"],
"spec_detail": labels["spec_detail"],
"brand_name": bname,
"qty_by_unit": {},
"rows": 0,
"_projects": {},
},
)
if ucode is not None:
mat["qty_by_unit"][ucode] = round(mat["qty_by_unit"].get(ucode, 0.0) + q, 4)
@@ -327,6 +441,7 @@ async def dead_stock(
material_id=None,
brand_id=None,
specs_hash=None,
material_family=None,
limit: int = 200,
) -> dict:
"""Stock rows with quantity > 0 and NO ledger activity within `idle_days`.
@@ -370,6 +485,10 @@ async def dead_stock(
stmt = _stock_filters(
stmt, material_id=material_id, brand_id=brand_id, specs_hash=specs_hash
)
if material_family:
stmt = stmt.where(
InventoryItem.material_id.in_(await _family_material_ids(db, material_family))
)
stmt = _scope_filters(stmt, project_ids=project_ids, project_id=project_id)
rows = (await db.execute(stmt)).all()
@@ -406,6 +525,7 @@ async def daily_flow(
material_id=None,
brand_id=None,
specs_hash=None,
material_family=None,
) -> dict:
"""Per-day buckets from the signed ledger.
@@ -440,6 +560,10 @@ async def daily_flow(
if warehouse_id is not None:
scope_conditions.append(InventoryTransaction.warehouse_id == warehouse_id)
what_conditions = []
family_ids: list[uuid.UUID] = []
if material_family:
family_ids = await _family_material_ids(db, material_family)
what_conditions.append(InventoryItem.material_id.in_(family_ids))
if material_id is not None:
what_conditions.append(InventoryItem.material_id == material_id)
if brand_id is not None:
@@ -525,13 +649,14 @@ async def daily_flow(
})
unit = None
if material_id is not None:
unit_material_id = material_id or (family_ids[0] if family_ids else None)
if unit_material_id is not None:
unit_row = (
await db.execute(
select(Unit.code, Unit.name_fa)
.select_from(Material)
.join(Unit, Unit.id == Material.canonical_unit_id)
.where(Material.id == material_id)
.where(Material.id == unit_material_id)
)
).first()
if unit_row is not None:
@@ -571,8 +696,14 @@ async def daily_control(
day: date_cls,
warehouse_id=None,
) -> dict:
"""One day (Tehran calendar): outbound / return / waste / purchase ticket
counts + quantities, the exact performer names, and the operation detail."""
"""One day (Tehran calendar): operational counts per category, the people
involved (warehouse clerk + site recipient) and the transaction feed.
Quantities are never exposed as a unitless total: every category carries a
per-unit breakdown (`units`) plus the number of distinct stock items
(`items`), so the UI can render «۱۲۲ عدد» for single-unit days and
«چندواحدی (۳ قلم کالا)» for mixed-unit days.
"""
day_start = datetime.combine(day, time.min, tzinfo=TEHRAN)
day_end = day_start + timedelta(days=1)
@@ -584,44 +715,88 @@ async def daily_control(
if warehouse_id is not None:
scope_conditions.append(InventoryTransaction.warehouse_id == warehouse_id)
count_rows = (
# ۱) شمارش/کمیت هر دسته، تفکیک‌شده بر اساس واحد + تعداد قلم کالای متمایز
category_unit_rows = (
await db.execute(
select(
InventoryTransaction.transaction_type,
Unit.code,
Unit.name_fa,
func.count().label("tickets"),
func.count(func.distinct(InventoryItem.id)).label("items"),
func.sum(func.abs(InventoryTransaction.quantity)).label("qty"),
)
.select_from(InventoryTransaction)
.join(Warehouse, Warehouse.id == InventoryTransaction.warehouse_id)
.join(InventoryItem, InventoryItem.id == InventoryTransaction.inventory_id)
.join(Unit, Unit.id == InventoryItem.unit_id)
.where(
InventoryTransaction.transaction_date >= day_start,
InventoryTransaction.transaction_date < day_end,
InventoryTransaction.transaction_type.in_(OPS_TYPES),
*scope_conditions,
)
.group_by(InventoryTransaction.transaction_type)
)
).all()
by_type = {str(tx_type): {"tickets": tickets, "qty": _qty(qty)} for tx_type, tickets, qty in count_rows}
performers_rows = (
await db.execute(
select(
InventoryTransaction.performed_by,
func.count().label("tickets"),
func.sum(func.abs(InventoryTransaction.quantity)).label("qty"),
)
.select_from(InventoryTransaction)
.join(Warehouse, Warehouse.id == InventoryTransaction.warehouse_id)
.where(
InventoryTransaction.transaction_date >= day_start,
InventoryTransaction.transaction_date < day_end,
*scope_conditions,
)
.group_by(InventoryTransaction.performed_by)
.order_by(func.count().desc())
.group_by(InventoryTransaction.transaction_type, Unit.code, Unit.name_fa)
)
).all()
categories: dict[str, dict] = {
key: {"tickets": 0, "items": 0, "qty": 0.0, "units": []} for key in OPS_CATEGORIES
}
for tx_type, unit_code, unit_name_fa, tickets, items, qty in category_unit_rows:
bucket = categories[_ops_category(str(tx_type))]
bucket["tickets"] += tickets
bucket["items"] += items
bucket["qty"] = round(bucket["qty"] + _qty(qty), 4)
bucket["units"].append({"code": unit_code, "name_fa": unit_name_fa, "qty": _qty(qty)})
# ۲) اشخاص درگیر: ثبت‌کننده‌ی انبار (performed_by) و تحویل‌گیرنده (delivered_to)
async def _people(field, role: str) -> list[dict]:
rows = (
await db.execute(
select(
field,
InventoryTransaction.transaction_type,
func.count(),
func.sum(func.abs(InventoryTransaction.quantity)),
)
.select_from(InventoryTransaction)
.join(Warehouse, Warehouse.id == InventoryTransaction.warehouse_id)
.where(
InventoryTransaction.transaction_date >= day_start,
InventoryTransaction.transaction_date < day_end,
InventoryTransaction.transaction_type.in_(OPS_TYPES),
field.isnot(None),
*scope_conditions,
)
.group_by(field, InventoryTransaction.transaction_type)
)
).all()
people: dict[str, dict] = {}
for name, tx_type, count, qty in rows:
clean = (name or "").strip()
if not clean:
continue
person = people.setdefault(
clean,
{
"name": clean,
"role": role,
"tickets": 0,
"qty": 0.0,
"categories": {key: 0 for key in OPS_CATEGORIES},
},
)
person["tickets"] += count
person["qty"] = round(person["qty"] + _qty(qty), 4)
person["categories"][_ops_category(str(tx_type))] += count
return list(people.values())
people = await _people(InventoryTransaction.performed_by, "performer")
people += await _people(InventoryTransaction.delivered_to, "recipient")
people.sort(key=lambda p: (p["role"] != "performer", -p["tickets"], p["name"]))
# ۳) ریز عملیات امروز (همراه با انبار، برند، واحد فارسی و تحویل‌گیرنده)
detail_rows = (
await db.execute(
select(
@@ -630,18 +805,19 @@ async def daily_control(
Material.name,
InventoryItem.specs,
Unit.code,
Unit.name_fa,
Brand.name,
)
.select_from(InventoryTransaction)
.join(Warehouse, Warehouse.id == InventoryTransaction.warehouse_id)
.join(InventoryItem, InventoryItem.id == InventoryTransaction.inventory_id)
.join(Material, Material.id == InventoryItem.material_id)
.join(Unit, Unit.id == InventoryItem.unit_id)
.outerjoin(Brand, Brand.id == InventoryItem.brand_id)
.where(
InventoryTransaction.transaction_date >= day_start,
InventoryTransaction.transaction_date < day_end,
InventoryTransaction.transaction_type.in_(
("WITHDRAWAL", "RETURN", "DAMAGE", "WASTE", "LOSS", "PURCHASE")
),
InventoryTransaction.transaction_type.in_(OPS_TYPES),
*scope_conditions,
)
.order_by(InventoryTransaction.transaction_date.desc())
@@ -649,15 +825,17 @@ async def daily_control(
)
).all()
zero = {"tickets": 0, "qty": 0.0}
items = []
for tx, wh_name, mat_name, specs, ucode in detail_rows:
for tx, wh_name, mat_name, specs, unit_code, unit_name_fa, brand_name in detail_rows:
items.append({
"id": str(tx.id),
"transaction_type": str(tx.transaction_type),
"quantity": _qty(tx.quantity),
"unit_code": ucode,
"unit_code": unit_code,
"unit_name_fa": unit_name_fa,
"performed_by": tx.performed_by,
"delivered_to": tx.delivered_to,
"brand": brand_name,
"note": tx.note,
"warehouse": wh_name,
"material": mat_name,
@@ -668,18 +846,20 @@ async def daily_control(
return {
"data": {
"date": day.isoformat(),
"outbound": by_type.get("WITHDRAWAL", zero),
"returns": by_type.get("RETURN", zero),
"waste": {
"tickets": sum(by_type.get(t, zero)["tickets"] for t in WASTE_TYPES),
"qty": round(sum(by_type.get(t, zero)["qty"] for t in WASTE_TYPES), 4),
},
"purchases": by_type.get("PURCHASE", zero),
"performers": [
{"name": name or "نامشخص", "tickets": tickets, "qty": _qty(qty)}
for name, tickets, qty in performers_rows
],
"outbound": categories["outbound"],
"returns": categories["returns"],
"waste": categories["waste"],
"purchases": categories["purchases"],
# همیشه آرایه‌اند (هرگز null/حذف‌شده) تا مصرف‌کننده‌ها امن بمانند
"people": people,
"items": items,
# سازگاری عقب‌رو: باندل‌های فرانت که هنوز فقط performers را می‌خوانند
# با API جدید کرش نکنند (تا زمان انتشار باندل تازه)
"performers": [
{"name": person["name"], "tickets": person["tickets"], "qty": person["qty"]}
for person in people
if person["role"] == "performer"
],
}
}
backend/app/services/material_naming.py
+121
View File
@@ -0,0 +1,121 @@
"""نام‌گذاری متریال — تفکیک «خانواده» از «مشخصه/واریانت» به‌صورت عمومی.
برخی ردیف‌های کاتالوگ، سایز/گرید را داخل نام ماده دارند
(«بلوک سیمانی ۲۰×۲۰×۴۰»، «لوله پلی‌اتیلن PE100 سایز ۲۵»). این ماژول با قواعد
الگویی (ابعاد NxNxN، کلیدواژه‌های مشخصه، کدهای گرید لاتین) نام را به
«خانواده» و «واریانت» می‌شکند — بدون هیچ هاردکدِ نام محصول مشخصی.
"""
import re
_DIGIT_MAP = str.maketrans("۰۱۲۳۴۵۶۷۸۹٠١٢٣٤٥٦٧٨٩", "01234567890123456789")
# واژه‌هایی که شروع بخش مشخصه را نشان می‌دهند (ابعاد/درجه/تیپ/مدل/…)
_ATTRIBUTE_WORDS = {
"سایز",
"تیپ",
"مدل",
"درجه",
"ضخامت",
"قطر",
"ابعاد",
"طول",
"عرض",
"ارتفاع",
"رنگ",
"وزن",
"ظرفیت",
"شماره",
"کلاس",
}
# ابعاد ترکیبی: ۲۰×۲۰×۴۰ / ۲×۲٫۵ / 60x60 / 2*2.5
_DIMENSION_RE = re.compile(r"^\d+(?:[×xX*]\d+(?:[.,٫/]\d+)?)+$")
# کد گرید: PE100 / PN10 / DN50 / A3 / SN4 / ST37
_GRADE_RE = re.compile(r"^(?:PN|PE|DN|SN|SCH|ST|A|B|C)[-]?\d+$")
_NUMBER_RE = re.compile(r"^\d+(?:[.,٫/]\d+)?$")
_LATIN_WORD_RE = re.compile(r"^[A-Za-z]{2,6}$")
STANDARD_VARIANT_LABEL = "استاندارد"
def normalize_digits(text: str) -> str:
return (text or "").translate(_DIGIT_MAP)
def _is_spec_start(tokens: list[str], index: int) -> bool:
token = tokens[index]
normalized = normalize_digits(token)
if normalized in _ATTRIBUTE_WORDS: # «سایز»، «تیپ»، «مدل»، …
return True
if _DIMENSION_RE.match(normalized): # ۲۰×۲۰×۴۰
return True
if _GRADE_RE.match(token): # PE100 / PN10 / A3
return True
# کد لاتین کوتاه که پشتش عدد می‌آید: «IPE ۱۴»
if (
_LATIN_WORD_RE.match(token)
and index + 1 < len(tokens)
and _NUMBER_RE.match(normalize_digits(tokens[index + 1]))
):
return True
# عددِ تنها در انتهای نام (بدون کلیدواژه): «لوله مسی ۱۵»
if _NUMBER_RE.match(normalized) and index == len(tokens) - 1 and index >= 1:
return True
return False
def split_material_name(name: str) -> tuple[str, str | None]:
"""«لوله پلی‌اتیلن PE100 سایز ۲۵» → («لوله پلی‌اتیلن», «PE100 سایز ۲۵»)."""
tokens = [token for token in (name or "").split() if token]
for index in range(1, len(tokens)):
if _is_spec_start(tokens, index):
family = " ".join(tokens[:index]).strip()
variant = " ".join(tokens[index:]).strip()
if family and variant:
return family, variant
return " ".join(tokens).strip(), None
def material_family(name: str) -> str:
"""نام خانواده (بدون واریانت)."""
family, _ = split_material_name(name)
return " ".join(family.split())
def material_family_key(name: str) -> str:
"""شناسه‌ی پایدار خانواده — همان نام نرمال‌شده (بدون UUID اختصاصی)."""
return material_family(name)
def variant_label(name: str) -> str:
"""برچسب واریانت؛ در نبود واریانت، «استاندارد»."""
_, variant = split_material_name(name)
return variant or STANDARD_VARIANT_LABEL
def spec_label_from_specs(specs: dict | None) -> str | None:
"""برچسب کوتاه مشخصات از JSON موجودی (فقط مقدارها، نه «کلید: مقدار»)."""
if not specs:
return None
values = [str(value).strip() for value in specs.values() if str(value).strip()]
return "، ".join(values) if values else None
def material_entry_labels(material_name: str, specs: dict | None) -> dict:
"""برچسب‌های نمایشی یک ردیف کاتالوگ: خانواده، واریانت، مشخصه.
`spec_label` (برچسب محور X) = مقدار مشخصات JSON اگر موجود باشد، وگرنه
واریانتِ استخراج‌شده از نام، وگرنه «استاندارد».
"""
family = material_family(material_name)
variant = variant_label(material_name)
spec_from_json = spec_label_from_specs(specs)
# برچسب محور X: واریانت (سایز/گرید) ترجیح دارد؛ در نبود آن مقدار مشخصات
axis_label = variant if variant != STANDARD_VARIANT_LABEL else (spec_from_json or STANDARD_VARIANT_LABEL)
return {
"family_id": material_family_key(material_name),
"family_name": family,
"variant_label": variant,
"spec_label": axis_label,
"spec_detail": spec_from_json,
}