Add Phase 21 stock item master mapping intelligence

This commit is contained in:
A R R R Associates
2026-08-24 22:07:16 +05:30
parent b8fbaf55ec
commit 0fab925e71
11 changed files with 1777 additions and 10 deletions
+701
View File
@@ -0,0 +1,701 @@
from __future__ import annotations
import json
import re
from datetime import datetime, timezone
from difflib import SequenceMatcher
from typing import Any
from sqlalchemy import delete, func, or_, select
from app.modules.accounting.stock_models import (
AccountingStockGroup,
AccountingStockItem,
AccountingStockItemMapping,
AccountingStockSyncRun,
AccountingStockUnit,
AccountingStockUnitMapping,
)
from app.modules.accounting.voucher_engine_models import AccountingVoucherItemLine
def _utcnow():
return datetime.now(timezone.utc)
def _s(value: Any) -> str:
return str(value or "").strip()
def normalize_text(value: Any) -> str:
return re.sub(r"[^a-z0-9]+", " ", _s(value).casefold()).strip()
def normalize_unit(value: Any) -> str:
text = normalize_text(value)
aliases = {
"kgs": "kg",
"kilogram": "kg",
"kilograms": "kg",
"nos": "nos",
"no": "nos",
"numbers": "nos",
"pcs": "pcs",
"pieces": "pcs",
"piece": "pcs",
"mtrs": "mtr",
"meters": "mtr",
"metres": "mtr",
"ton": "mt",
"tons": "mt",
"tonne": "mt",
"tonnes": "mt",
}
return aliases.get(text, text)
def _payload(row):
try:
return json.loads(row or "{}")
except Exception:
return {}
def persist_stock_snapshot(
db,
*,
tenant_id: int,
client_id: int,
tally_guid: str,
company_name: str,
stock_groups: list[dict],
stock_items: list[dict],
units: list[dict],
user_id: int,
):
tally_guid = _s(tally_guid)
if not tally_guid:
raise ValueError("Tally company GUID is required.")
db.execute(
delete(AccountingStockGroup).where(
AccountingStockGroup.tenant_id == int(tenant_id),
AccountingStockGroup.client_id == int(client_id),
AccountingStockGroup.tally_guid == tally_guid,
)
)
db.execute(
delete(AccountingStockItem).where(
AccountingStockItem.tenant_id == int(tenant_id),
AccountingStockItem.client_id == int(client_id),
AccountingStockItem.tally_guid == tally_guid,
)
)
db.execute(
delete(AccountingStockUnit).where(
AccountingStockUnit.tenant_id == int(tenant_id),
AccountingStockUnit.client_id == int(client_id),
AccountingStockUnit.tally_guid == tally_guid,
)
)
db.flush()
synced = _utcnow()
for row in stock_groups or []:
name = _s(row.get("name"))
if not name:
continue
db.add(
AccountingStockGroup(
tenant_id=int(tenant_id),
client_id=int(client_id),
tally_guid=tally_guid,
company_name=company_name,
master_guid=_s(row.get("guid")),
name=name,
parent_name=_s(row.get("parent")),
base_units=_s(row.get("base_units")),
source_payload_json=json.dumps(row, ensure_ascii=False, default=str),
synced_at_utc=synced,
)
)
for row in stock_items or []:
name = _s(row.get("name"))
if not name:
continue
db.add(
AccountingStockItem(
tenant_id=int(tenant_id),
client_id=int(client_id),
tally_guid=tally_guid,
company_name=company_name,
master_guid=_s(row.get("guid")),
name=name,
parent_group_name=_s(row.get("parent")),
category_name=_s(row.get("category")),
base_units=_s(row.get("base_units")),
additional_units=_s(row.get("additional_units")),
hsn_code=_s(row.get("hsn_code")),
gst_applicable=_s(row.get("gst_applicable")),
gst_type_of_supply=_s(row.get("gst_type_of_supply")),
opening_balance=float(row.get("opening_balance") or 0),
opening_value=float(row.get("opening_value") or 0),
opening_rate=_s(row.get("opening_rate")),
normalized_name=normalize_text(name),
source_payload_json=json.dumps(row, ensure_ascii=False, default=str),
synced_at_utc=synced,
)
)
for row in units or []:
name = _s(row.get("name"))
if not name:
continue
db.add(
AccountingStockUnit(
tenant_id=int(tenant_id),
client_id=int(client_id),
tally_guid=tally_guid,
company_name=company_name,
master_guid=_s(row.get("guid")),
name=name,
original_name=_s(row.get("original_name")),
is_simple_unit=_s(row.get("is_simple_unit")),
base_units=_s(row.get("base_units")),
additional_units=_s(row.get("additional_units")),
conversion=_s(row.get("conversion")),
normalized_name=normalize_unit(name),
source_payload_json=json.dumps(row, ensure_ascii=False, default=str),
synced_at_utc=synced,
)
)
run = AccountingStockSyncRun(
tenant_id=int(tenant_id),
client_id=int(client_id),
tally_guid=tally_guid,
company_name=company_name,
status="completed",
stock_group_count=len([x for x in stock_groups or [] if _s(x.get("name"))]),
stock_item_count=len([x for x in stock_items or [] if _s(x.get("name"))]),
unit_count=len([x for x in units or [] if _s(x.get("name"))]),
requested_by_user_id=int(user_id),
)
db.add(run)
db.commit()
db.refresh(run)
return run
def list_stock_items(
db,
*,
tenant_id: int,
client_id: int,
tally_guid: str,
q: str = "",
group: str = "",
hsn: str = "",
page: int = 1,
per_page: int = 50,
):
stmt = select(AccountingStockItem).where(
AccountingStockItem.tenant_id == int(tenant_id),
AccountingStockItem.client_id == int(client_id),
AccountingStockItem.tally_guid == _s(tally_guid),
)
if _s(q):
like = f"%{_s(q)}%"
stmt = stmt.where(
or_(
AccountingStockItem.name.ilike(like),
AccountingStockItem.parent_group_name.ilike(like),
AccountingStockItem.hsn_code.ilike(like),
)
)
if _s(group):
stmt = stmt.where(AccountingStockItem.parent_group_name == _s(group))
if _s(hsn):
stmt = stmt.where(AccountingStockItem.hsn_code == _s(hsn))
count_stmt = select(func.count()).select_from(stmt.subquery())
total = int(db.execute(count_stmt).scalar_one() or 0)
per_page = max(10, min(200, int(per_page or 50)))
pages = max(1, (total + per_page - 1) // per_page)
page = max(1, min(int(page or 1), pages))
rows = list(
db.execute(
stmt.order_by(
AccountingStockItem.parent_group_name,
AccountingStockItem.name,
)
.offset((page - 1) * per_page)
.limit(per_page)
).scalars().all()
)
return rows, total, page, pages
def list_stock_groups(db, *, tenant_id: int, client_id: int, tally_guid: str):
return list(
db.execute(
select(AccountingStockGroup)
.where(
AccountingStockGroup.tenant_id == int(tenant_id),
AccountingStockGroup.client_id == int(client_id),
AccountingStockGroup.tally_guid == _s(tally_guid),
)
.order_by(AccountingStockGroup.parent_name, AccountingStockGroup.name)
).scalars().all()
)
def list_units(db, *, tenant_id: int, client_id: int, tally_guid: str):
return list(
db.execute(
select(AccountingStockUnit)
.where(
AccountingStockUnit.tenant_id == int(tenant_id),
AccountingStockUnit.client_id == int(client_id),
AccountingStockUnit.tally_guid == _s(tally_guid),
)
.order_by(AccountingStockUnit.name)
).scalars().all()
)
def save_stock_mapping(
db,
*,
tenant_id: int,
client_id: int,
tally_guid: str,
source_type: str,
party_key: str,
source_description: str,
hsn_sac: str,
stock_item_id: int,
user_id: int,
note: str = "",
):
item = db.get(AccountingStockItem, int(stock_item_id))
if (
not item
or int(item.tenant_id) != int(tenant_id)
or int(item.client_id) != int(client_id)
or item.tally_guid != _s(tally_guid)
):
raise ValueError("Selected Tally stock item is not valid for this client/company.")
source_type = _s(source_type).lower() or "generic"
party_key = normalize_text(party_key)
source_description = _s(source_description)
source_norm = normalize_text(source_description)
hsn_sac = _s(hsn_sac)
if not source_norm:
raise ValueError("Source item description is required.")
mapping = db.execute(
select(AccountingStockItemMapping).where(
AccountingStockItemMapping.tenant_id == int(tenant_id),
AccountingStockItemMapping.client_id == int(client_id),
AccountingStockItemMapping.tally_guid == _s(tally_guid),
AccountingStockItemMapping.source_type == source_type,
AccountingStockItemMapping.party_key == party_key,
AccountingStockItemMapping.source_description_norm == source_norm,
AccountingStockItemMapping.hsn_sac == hsn_sac,
)
).scalar_one_or_none()
if mapping is None:
mapping = AccountingStockItemMapping(
tenant_id=int(tenant_id),
client_id=int(client_id),
tally_guid=_s(tally_guid),
source_type=source_type,
party_key=party_key,
source_description=source_description,
source_description_norm=source_norm,
hsn_sac=hsn_sac,
stock_item_id=item.id,
confidence=100,
mapping_origin="manual",
created_by_user_id=int(user_id),
)
mapping.source_description = source_description
mapping.stock_item_id = item.id
mapping.confidence = 100
mapping.mapping_origin = "manual"
mapping.note = _s(note)
mapping.updated_by_user_id = int(user_id)
mapping.updated_at_utc = _utcnow()
db.add(mapping)
db.commit()
db.refresh(mapping)
return mapping
def save_unit_mapping(
db,
*,
tenant_id: int,
client_id: int,
tally_guid: str,
source_unit: str,
target_unit_id: int,
quantity_factor: float,
user_id: int,
note: str = "",
):
unit = db.get(AccountingStockUnit, int(target_unit_id))
if (
not unit
or int(unit.tenant_id) != int(tenant_id)
or int(unit.client_id) != int(client_id)
or unit.tally_guid != _s(tally_guid)
):
raise ValueError("Selected Tally unit is not valid for this client/company.")
source_unit = _s(source_unit)
source_norm = normalize_unit(source_unit)
factor = float(quantity_factor or 0)
if not source_norm:
raise ValueError("Source unit is required.")
if factor <= 0:
raise ValueError("Quantity conversion factor must be greater than zero.")
mapping = db.execute(
select(AccountingStockUnitMapping).where(
AccountingStockUnitMapping.tenant_id == int(tenant_id),
AccountingStockUnitMapping.client_id == int(client_id),
AccountingStockUnitMapping.tally_guid == _s(tally_guid),
AccountingStockUnitMapping.source_unit_norm == source_norm,
)
).scalar_one_or_none()
if mapping is None:
mapping = AccountingStockUnitMapping(
tenant_id=int(tenant_id),
client_id=int(client_id),
tally_guid=_s(tally_guid),
source_unit=source_unit,
source_unit_norm=source_norm,
target_unit_id=unit.id,
quantity_factor=factor,
created_by_user_id=int(user_id),
)
mapping.source_unit = source_unit
mapping.target_unit_id = unit.id
mapping.quantity_factor = factor
mapping.note = _s(note)
mapping.updated_at_utc = _utcnow()
db.add(mapping)
db.commit()
db.refresh(mapping)
return mapping
def resolve_unit(
db,
*,
tenant_id: int,
client_id: int,
tally_guid: str,
source_unit: str,
):
source_norm = normalize_unit(source_unit)
if not source_norm:
return {"matched": True, "target_unit": "", "quantity_factor": 1.0, "reason": "No source unit."}
mapping = db.execute(
select(AccountingStockUnitMapping, AccountingStockUnit)
.join(AccountingStockUnit, AccountingStockUnit.id == AccountingStockUnitMapping.target_unit_id)
.where(
AccountingStockUnitMapping.tenant_id == int(tenant_id),
AccountingStockUnitMapping.client_id == int(client_id),
AccountingStockUnitMapping.tally_guid == _s(tally_guid),
AccountingStockUnitMapping.source_unit_norm == source_norm,
)
).first()
if mapping:
m, unit = mapping
return {
"matched": True,
"target_unit": unit.name,
"quantity_factor": float(m.quantity_factor or 1),
"reason": "Learned unit mapping.",
}
units = list_units(db, tenant_id=tenant_id, client_id=client_id, tally_guid=tally_guid)
exact = [u for u in units if normalize_unit(u.name) == source_norm or normalize_unit(u.original_name) == source_norm]
if len(exact) == 1:
return {
"matched": True,
"target_unit": exact[0].name,
"quantity_factor": 1.0,
"reason": "Exact normalized Tally unit.",
}
return {
"matched": False,
"target_unit": "",
"quantity_factor": 1.0,
"reason": f"Unit '{source_unit}' is not mapped uniquely.",
}
def resolve_stock_item(
db,
*,
tenant_id: int,
client_id: int,
tally_guid: str,
source_type: str,
party_key: str,
source_description: str,
hsn_sac: str = "",
):
source_type = _s(source_type).lower() or "generic"
source_norm = normalize_text(source_description)
party_norm = normalize_text(party_key)
hsn_sac = _s(hsn_sac)
if not source_norm:
return {"matched": False, "confidence": 0, "reason": "Source item description is empty.", "candidates": []}
learned = list(
db.execute(
select(AccountingStockItemMapping, AccountingStockItem)
.join(AccountingStockItem, AccountingStockItem.id == AccountingStockItemMapping.stock_item_id)
.where(
AccountingStockItemMapping.tenant_id == int(tenant_id),
AccountingStockItemMapping.client_id == int(client_id),
AccountingStockItemMapping.tally_guid == _s(tally_guid),
AccountingStockItemMapping.source_description_norm == source_norm,
AccountingStockItemMapping.source_type.in_([source_type, "generic"]),
)
.order_by(AccountingStockItemMapping.confidence.desc())
).all()
)
ranked_learned = []
for mapping, item in learned:
score = int(mapping.confidence or 100)
if mapping.party_key and mapping.party_key == party_norm:
score += 20
elif mapping.party_key:
score -= 30
if hsn_sac and mapping.hsn_sac and mapping.hsn_sac == hsn_sac:
score += 10
elif hsn_sac and mapping.hsn_sac and mapping.hsn_sac != hsn_sac:
score -= 20
ranked_learned.append((score, item, mapping))
ranked_learned.sort(key=lambda row: (-row[0], row[1].name))
if ranked_learned and ranked_learned[0][0] >= 90:
score, item, mapping = ranked_learned[0]
return {
"matched": True,
"stock_item": item,
"confidence": min(100, score),
"reason": "Confirmed client/source stock-item mapping.",
"mapping_id": mapping.id,
"candidates": [],
}
items = list(
db.execute(
select(AccountingStockItem).where(
AccountingStockItem.tenant_id == int(tenant_id),
AccountingStockItem.client_id == int(client_id),
AccountingStockItem.tally_guid == _s(tally_guid),
)
).scalars().all()
)
exact = [item for item in items if item.normalized_name == source_norm]
if len(exact) == 1:
return {
"matched": True,
"stock_item": exact[0],
"confidence": 98,
"reason": "Exact normalized Tally stock-item name.",
"candidates": [],
}
scored = []
source_tokens = {t for t in source_norm.split() if len(t) >= 2}
for item in items:
item_tokens = {t for t in item.normalized_name.split() if len(t) >= 2}
common = source_tokens & item_tokens
token_score = 0
if source_tokens:
token_score = int(round((len(common) / len(source_tokens)) * 70))
similarity = int(round(SequenceMatcher(None, source_norm, item.normalized_name).ratio() * 25))
hsn_bonus = 0
if hsn_sac and item.hsn_code:
if hsn_sac == item.hsn_code:
hsn_bonus = 20
elif hsn_sac[:4] and item.hsn_code[:4] == hsn_sac[:4]:
hsn_bonus = 10
else:
hsn_bonus = -20
score = max(0, min(100, token_score + similarity + hsn_bonus))
if score >= 45:
scored.append((score, item))
scored.sort(key=lambda row: (-row[0], row[1].name))
candidates = [
{
"id": item.id,
"name": item.name,
"group": item.parent_group_name,
"hsn": item.hsn_code,
"unit": item.base_units,
"confidence": score,
}
for score, item in scored[:8]
]
if scored:
top_score = scored[0][0]
same_top = [row for row in scored if row[0] == top_score]
second = scored[1][0] if len(scored) > 1 else 0
if top_score >= 88 and len(same_top) == 1 and (top_score - second >= 10):
return {
"matched": True,
"stock_item": scored[0][1],
"confidence": top_score,
"reason": "Unique conservative HSN/name match.",
"candidates": candidates,
}
return {
"matched": False,
"confidence": scored[0][0] if scored else 0,
"reason": "No sufficiently unique stock-item mapping. Reviewer mapping is required.",
"candidates": candidates,
}
def list_mappings(db, *, tenant_id: int, client_id: int, tally_guid: str):
return list(
db.execute(
select(AccountingStockItemMapping, AccountingStockItem)
.join(AccountingStockItem, AccountingStockItem.id == AccountingStockItemMapping.stock_item_id)
.where(
AccountingStockItemMapping.tenant_id == int(tenant_id),
AccountingStockItemMapping.client_id == int(client_id),
AccountingStockItemMapping.tally_guid == _s(tally_guid),
)
.order_by(
AccountingStockItemMapping.source_type,
AccountingStockItemMapping.source_description,
)
).all()
)
def list_unit_mappings(db, *, tenant_id: int, client_id: int, tally_guid: str):
return list(
db.execute(
select(AccountingStockUnitMapping, AccountingStockUnit)
.join(AccountingStockUnit, AccountingStockUnit.id == AccountingStockUnitMapping.target_unit_id)
.where(
AccountingStockUnitMapping.tenant_id == int(tenant_id),
AccountingStockUnitMapping.client_id == int(client_id),
AccountingStockUnitMapping.tally_guid == _s(tally_guid),
)
.order_by(AccountingStockUnitMapping.source_unit)
).all()
)
def needs_mapping(
db,
*,
tenant_id: int,
client_id: int,
tally_guid: str,
limit: int = 200,
):
# Reuse already-generated voucher source evidence. This avoids creating
# another staging table and works with Phase 18 E-Way Bill drafts today.
rows = list(
db.execute(
select(AccountingVoucherItemLine)
.join(AccountingVoucherItemLine.draft)
.where(
AccountingVoucherItemLine.draft.has(
tenant_id=int(tenant_id),
client_id=int(client_id),
tally_guid=_s(tally_guid),
)
)
.order_by(AccountingVoucherItemLine.id.desc())
.limit(max(1, min(1000, int(limit))))
).scalars().all()
)
result = []
seen = set()
for line in rows:
payload = _payload(line.source_item_json)
description = _s(payload.get("item_name") or payload.get("product_name") or line.stock_item_name)
hsn = _s(payload.get("hsn") or payload.get("hsn_code") or line.hsn_sac)
party = _s(payload.get("party_name") or payload.get("supplier_name") or payload.get("customer_name"))
source_type = _s(payload.get("source_type") or "ewaybill").lower()
key = (source_type, normalize_text(party), normalize_text(description), hsn)
if not description or key in seen:
continue
seen.add(key)
resolved = resolve_stock_item(
db,
tenant_id=tenant_id,
client_id=client_id,
tally_guid=tally_guid,
source_type=source_type,
party_key=party,
source_description=description,
hsn_sac=hsn,
)
if resolved.get("matched"):
continue
result.append(
{
"source_type": source_type,
"party": party,
"description": description,
"hsn": hsn,
"source_unit": _s(payload.get("unit")),
"confidence": int(resolved.get("confidence") or 0),
"reason": resolved.get("reason") or "",
"candidates": resolved.get("candidates") or [],
}
)
return result[:limit]
def stock_summary(db, *, tenant_id: int, client_id: int, tally_guid: str):
def count(model):
return int(
db.execute(
select(func.count(model.id)).where(
model.tenant_id == int(tenant_id),
model.client_id == int(client_id),
model.tally_guid == _s(tally_guid),
)
).scalar_one() or 0
)
return {
"groups": count(AccountingStockGroup),
"items": count(AccountingStockItem),
"units": count(AccountingStockUnit),
"mappings": count(AccountingStockItemMapping),
"unit_mappings": count(AccountingStockUnitMapping),
}