Add Phase 18 automatic stock item voucher preparation
This commit is contained in:
A R R R Associates
@@ -0,0 +1,370 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from datetime import date, datetime
|
||||
from decimal import Decimal
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy import MetaData, Table, and_, inspect, select
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SourceDescriptor:
|
||||
kind: str
|
||||
table_name: str
|
||||
score: int
|
||||
columns: tuple[str, ...]
|
||||
|
||||
|
||||
BANK_TABLE_HINTS = (
|
||||
"bank_statement_transaction",
|
||||
"bank_transactions",
|
||||
"bank_analyzer_transaction",
|
||||
"bank_statement_rows",
|
||||
"bank_entries",
|
||||
)
|
||||
|
||||
EWAY_TABLE_HINTS = (
|
||||
"ewaybill",
|
||||
"eway_bill",
|
||||
"e_way_bill",
|
||||
"accounting_purchase_source",
|
||||
"purchase_source",
|
||||
"sales_source",
|
||||
)
|
||||
|
||||
BANK_COLUMNS = {
|
||||
"id": ("id", "transaction_id", "row_id"),
|
||||
"client_id": ("client_id",),
|
||||
"date": ("transaction_date", "txn_date", "date", "value_date"),
|
||||
"narration": ("narration", "description", "particulars", "remarks"),
|
||||
"debit": ("debit", "withdrawal", "debit_amount"),
|
||||
"credit": ("credit", "deposit", "credit_amount"),
|
||||
"amount": ("amount",),
|
||||
"dr_cr": ("dr_cr", "type", "nature"),
|
||||
"party": ("final_party", "manual_party", "auto_party", "party", "counterparty"),
|
||||
"category": ("category", "final_category", "nature"),
|
||||
"account_no": ("account_number", "bank_account_number", "account_no"),
|
||||
"bank": ("bank_name", "bank"),
|
||||
"reference": ("reference", "ref_no", "utr", "transaction_reference"),
|
||||
"contra": ("contra_match_id", "contra_pair_id", "is_contra"),
|
||||
}
|
||||
|
||||
EWAY_COLUMNS = {
|
||||
"id": ("id", "item_id", "row_id"),
|
||||
"client_id": ("client_id",),
|
||||
"direction": ("direction", "supply_type", "transaction_type", "inward_outward"),
|
||||
"ewb_no": ("eway_bill_no", "ewb_no", "ewaybill_no"),
|
||||
"invoice_no": ("invoice_no", "doc_no", "document_number", "bill_no"),
|
||||
"invoice_date": ("invoice_date", "doc_date", "document_date", "date"),
|
||||
"party_name": ("supplier_name", "customer_name", "party_name", "trade_name", "legal_name"),
|
||||
"party_gstin": ("supplier_gstin", "customer_gstin", "party_gstin", "gstin"),
|
||||
"item_name": ("product_name", "item_name", "description", "product_description"),
|
||||
"hsn": ("hsn", "hsn_code", "hsn_sac"),
|
||||
"quantity": ("quantity", "qty"),
|
||||
"unit": ("unit", "uqc"),
|
||||
"rate": ("rate", "unit_rate"),
|
||||
"taxable_value": ("taxable_value", "taxable_amount", "assessable_value", "amount"),
|
||||
"cgst": ("cgst", "cgst_amount"),
|
||||
"sgst": ("sgst", "sgst_amount"),
|
||||
"igst": ("igst", "igst_amount"),
|
||||
"cess": ("cess", "cess_amount"),
|
||||
"tax_rate": ("tax_rate", "gst_rate", "rate_percent"),
|
||||
"vehicle_no": ("vehicle_no", "vehicle_number"),
|
||||
}
|
||||
|
||||
|
||||
def _norm(value: Any) -> str:
|
||||
return str(value or "").strip()
|
||||
|
||||
|
||||
def _float(value: Any) -> float:
|
||||
if value in (None, ""):
|
||||
return 0.0
|
||||
if isinstance(value, Decimal):
|
||||
return float(value)
|
||||
try:
|
||||
return float(str(value).replace(",", ""))
|
||||
except Exception:
|
||||
return 0.0
|
||||
|
||||
|
||||
def _filter_date(column, value: str):
|
||||
text = _norm(value)
|
||||
if not text:
|
||||
return None
|
||||
type_name = column.type.__class__.__name__.lower()
|
||||
if "date" in type_name or "time" in type_name:
|
||||
try:
|
||||
return date.fromisoformat(text)
|
||||
except Exception:
|
||||
return text
|
||||
return text
|
||||
|
||||
|
||||
def _date_text(value: Any) -> str:
|
||||
if value in (None, ""):
|
||||
return ""
|
||||
if isinstance(value, datetime):
|
||||
return value.date().isoformat()
|
||||
if isinstance(value, date):
|
||||
return value.isoformat()
|
||||
text = _norm(value)
|
||||
for fmt in ("%Y-%m-%d", "%d/%m/%Y", "%d-%m-%Y", "%Y%m%d"):
|
||||
try:
|
||||
return datetime.strptime(text, fmt).date().isoformat()
|
||||
except Exception:
|
||||
pass
|
||||
return text[:10]
|
||||
|
||||
|
||||
def _resolve(columns: set[str], choices: tuple[str, ...]) -> str:
|
||||
lower = {c.lower(): c for c in columns}
|
||||
for candidate in choices:
|
||||
if candidate.lower() in lower:
|
||||
return lower[candidate.lower()]
|
||||
return ""
|
||||
|
||||
|
||||
def _table_score(name: str, cols: set[str], hints: tuple[str, ...], mapping: dict) -> int:
|
||||
score = 0
|
||||
lname = name.lower()
|
||||
if any(h in lname for h in hints):
|
||||
score += 12
|
||||
for choices in mapping.values():
|
||||
if _resolve(cols, choices):
|
||||
score += 1
|
||||
return score
|
||||
|
||||
|
||||
def discover_sources(db) -> dict[str, list[SourceDescriptor]]:
|
||||
inspector = inspect(db.get_bind())
|
||||
result = {"bank": [], "eway": []}
|
||||
|
||||
for table_name in inspector.get_table_names():
|
||||
try:
|
||||
cols = {row["name"] for row in inspector.get_columns(table_name)}
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
bank_score = _table_score(table_name, cols, BANK_TABLE_HINTS, BANK_COLUMNS)
|
||||
if bank_score >= 18 and (
|
||||
_resolve(cols, BANK_COLUMNS["narration"])
|
||||
and (
|
||||
_resolve(cols, BANK_COLUMNS["amount"])
|
||||
or _resolve(cols, BANK_COLUMNS["debit"])
|
||||
or _resolve(cols, BANK_COLUMNS["credit"])
|
||||
)
|
||||
):
|
||||
result["bank"].append(
|
||||
SourceDescriptor("bank", table_name, bank_score, tuple(sorted(cols)))
|
||||
)
|
||||
|
||||
eway_score = _table_score(table_name, cols, EWAY_TABLE_HINTS, EWAY_COLUMNS)
|
||||
if eway_score >= 17 and (
|
||||
_resolve(cols, EWAY_COLUMNS["ewb_no"])
|
||||
and _resolve(cols, EWAY_COLUMNS["item_name"])
|
||||
and _resolve(cols, EWAY_COLUMNS["taxable_value"])
|
||||
):
|
||||
result["eway"].append(
|
||||
SourceDescriptor("eway", table_name, eway_score, tuple(sorted(cols)))
|
||||
)
|
||||
|
||||
for key in result:
|
||||
result[key] = sorted(result[key], key=lambda row: (-row.score, row.table_name))
|
||||
return result
|
||||
|
||||
|
||||
def _table(db, table_name: str):
|
||||
metadata = MetaData()
|
||||
return Table(table_name, metadata, autoload_with=db.get_bind())
|
||||
|
||||
|
||||
def bank_rows(
|
||||
db,
|
||||
*,
|
||||
table_name: str,
|
||||
client_id: int,
|
||||
from_date: str = "",
|
||||
to_date: str = "",
|
||||
limit: int = 500,
|
||||
):
|
||||
table = _table(db, table_name)
|
||||
cols = set(table.c.keys())
|
||||
|
||||
cmap = {key: _resolve(cols, choices) for key, choices in BANK_COLUMNS.items()}
|
||||
conditions = []
|
||||
if cmap["client_id"]:
|
||||
conditions.append(table.c[cmap["client_id"]] == int(client_id))
|
||||
if from_date and cmap["date"]:
|
||||
conditions.append(table.c[cmap["date"]] >= _filter_date(table.c[cmap["date"]], from_date))
|
||||
if to_date and cmap["date"]:
|
||||
conditions.append(table.c[cmap["date"]] <= _filter_date(table.c[cmap["date"]], to_date))
|
||||
|
||||
stmt = select(table)
|
||||
if conditions:
|
||||
stmt = stmt.where(and_(*conditions))
|
||||
if cmap["date"]:
|
||||
stmt = stmt.order_by(table.c[cmap["date"]].desc())
|
||||
stmt = stmt.limit(max(1, min(2000, int(limit))))
|
||||
|
||||
rows = []
|
||||
for raw in db.execute(stmt).mappings():
|
||||
debit = _float(raw.get(cmap["debit"])) if cmap["debit"] else 0.0
|
||||
credit = _float(raw.get(cmap["credit"])) if cmap["credit"] else 0.0
|
||||
amount = _float(raw.get(cmap["amount"])) if cmap["amount"] else 0.0
|
||||
dr_cr = _norm(raw.get(cmap["dr_cr"])).upper() if cmap["dr_cr"] else ""
|
||||
|
||||
if debit > 0:
|
||||
direction = "DEBIT"
|
||||
value = debit
|
||||
elif credit > 0:
|
||||
direction = "CREDIT"
|
||||
value = credit
|
||||
elif dr_cr.startswith("D"):
|
||||
direction = "DEBIT"
|
||||
value = abs(amount)
|
||||
elif dr_cr.startswith("C"):
|
||||
direction = "CREDIT"
|
||||
value = abs(amount)
|
||||
elif amount < 0:
|
||||
direction = "DEBIT"
|
||||
value = abs(amount)
|
||||
else:
|
||||
direction = "CREDIT"
|
||||
value = abs(amount)
|
||||
|
||||
rows.append(
|
||||
{
|
||||
"source_table": table_name,
|
||||
"source_id": _norm(raw.get(cmap["id"])) if cmap["id"] else "",
|
||||
"date": _date_text(raw.get(cmap["date"])) if cmap["date"] else "",
|
||||
"narration": _norm(raw.get(cmap["narration"])) if cmap["narration"] else "",
|
||||
"party": _norm(raw.get(cmap["party"])) if cmap["party"] else "",
|
||||
"category": _norm(raw.get(cmap["category"])) if cmap["category"] else "",
|
||||
"bank_name": _norm(raw.get(cmap["bank"])) if cmap["bank"] else "",
|
||||
"account_number": _norm(raw.get(cmap["account_no"])) if cmap["account_no"] else "",
|
||||
"reference": _norm(raw.get(cmap["reference"])) if cmap["reference"] else "",
|
||||
"direction": direction,
|
||||
"amount": round(value, 2),
|
||||
"contra_hint": _norm(raw.get(cmap["contra"])) if cmap["contra"] else "",
|
||||
}
|
||||
)
|
||||
return rows
|
||||
|
||||
|
||||
def eway_rows(
|
||||
db,
|
||||
*,
|
||||
table_name: str,
|
||||
client_id: int,
|
||||
direction: str = "",
|
||||
from_date: str = "",
|
||||
to_date: str = "",
|
||||
limit: int = 2000,
|
||||
):
|
||||
table = _table(db, table_name)
|
||||
cols = set(table.c.keys())
|
||||
cmap = {key: _resolve(cols, choices) for key, choices in EWAY_COLUMNS.items()}
|
||||
|
||||
conditions = []
|
||||
if cmap["client_id"]:
|
||||
conditions.append(table.c[cmap["client_id"]] == int(client_id))
|
||||
if from_date and cmap["invoice_date"]:
|
||||
conditions.append(table.c[cmap["invoice_date"]] >= _filter_date(table.c[cmap["invoice_date"]], from_date))
|
||||
if to_date and cmap["invoice_date"]:
|
||||
conditions.append(table.c[cmap["invoice_date"]] <= _filter_date(table.c[cmap["invoice_date"]], to_date))
|
||||
|
||||
stmt = select(table)
|
||||
if conditions:
|
||||
stmt = stmt.where(and_(*conditions))
|
||||
if cmap["invoice_date"]:
|
||||
stmt = stmt.order_by(table.c[cmap["invoice_date"]].desc())
|
||||
stmt = stmt.limit(max(1, min(10000, int(limit))))
|
||||
|
||||
normalized_direction = _norm(direction).upper()
|
||||
rows = []
|
||||
for raw in db.execute(stmt).mappings():
|
||||
raw_direction = _norm(raw.get(cmap["direction"])).upper() if cmap["direction"] else ""
|
||||
detected = ""
|
||||
if raw_direction in {"I", "IN", "INWARD", "PURCHASE"} or "INWARD" in raw_direction or "PURCHASE" in raw_direction:
|
||||
detected = "PURCHASE"
|
||||
elif raw_direction in {"O", "OUT", "OUTWARD", "SALES"} or "OUTWARD" in raw_direction or "SALES" in raw_direction:
|
||||
detected = "SALES"
|
||||
|
||||
if normalized_direction and detected and detected != normalized_direction:
|
||||
continue
|
||||
|
||||
taxable = round(abs(_float(raw.get(cmap["taxable_value"]))), 2)
|
||||
qty = _float(raw.get(cmap["quantity"])) if cmap["quantity"] else 0.0
|
||||
rate = _float(raw.get(cmap["rate"])) if cmap["rate"] else 0.0
|
||||
if not rate and qty:
|
||||
rate = taxable / qty
|
||||
|
||||
rows.append(
|
||||
{
|
||||
"source_table": table_name,
|
||||
"source_id": _norm(raw.get(cmap["id"])) if cmap["id"] else "",
|
||||
"direction": detected or normalized_direction or "UNKNOWN",
|
||||
"eway_bill_no": _norm(raw.get(cmap["ewb_no"])) if cmap["ewb_no"] else "",
|
||||
"invoice_no": _norm(raw.get(cmap["invoice_no"])) if cmap["invoice_no"] else "",
|
||||
"invoice_date": _date_text(raw.get(cmap["invoice_date"])) if cmap["invoice_date"] else "",
|
||||
"party_name": _norm(raw.get(cmap["party_name"])) if cmap["party_name"] else "",
|
||||
"party_gstin": _norm(raw.get(cmap["party_gstin"])) if cmap["party_gstin"] else "",
|
||||
"item_name": _norm(raw.get(cmap["item_name"])) if cmap["item_name"] else "",
|
||||
"hsn": _norm(raw.get(cmap["hsn"])) if cmap["hsn"] else "",
|
||||
"quantity": qty,
|
||||
"unit": _norm(raw.get(cmap["unit"])) if cmap["unit"] else "",
|
||||
"rate": round(rate, 4),
|
||||
"taxable_value": taxable,
|
||||
"cgst": round(abs(_float(raw.get(cmap["cgst"]))), 2) if cmap["cgst"] else 0.0,
|
||||
"sgst": round(abs(_float(raw.get(cmap["sgst"]))), 2) if cmap["sgst"] else 0.0,
|
||||
"igst": round(abs(_float(raw.get(cmap["igst"]))), 2) if cmap["igst"] else 0.0,
|
||||
"cess": round(abs(_float(raw.get(cmap["cess"]))), 2) if cmap["cess"] else 0.0,
|
||||
"tax_rate": _float(raw.get(cmap["tax_rate"])) if cmap["tax_rate"] else 0.0,
|
||||
"vehicle_no": _norm(raw.get(cmap["vehicle_no"])) if cmap["vehicle_no"] else "",
|
||||
}
|
||||
)
|
||||
return rows
|
||||
|
||||
|
||||
def group_eway_invoices(rows: list[dict]) -> list[dict]:
|
||||
grouped: dict[tuple, dict] = {}
|
||||
for row in rows:
|
||||
key = (
|
||||
row.get("direction") or "",
|
||||
row.get("eway_bill_no") or "",
|
||||
row.get("invoice_no") or "",
|
||||
row.get("invoice_date") or "",
|
||||
row.get("party_gstin") or "",
|
||||
row.get("party_name") or "",
|
||||
)
|
||||
group = grouped.setdefault(
|
||||
key,
|
||||
{
|
||||
"direction": row.get("direction") or "",
|
||||
"eway_bill_no": row.get("eway_bill_no") or "",
|
||||
"invoice_no": row.get("invoice_no") or "",
|
||||
"invoice_date": row.get("invoice_date") or "",
|
||||
"party_name": row.get("party_name") or "",
|
||||
"party_gstin": row.get("party_gstin") or "",
|
||||
"source_table": row.get("source_table") or "",
|
||||
"source_ids": [],
|
||||
"items": [],
|
||||
"taxable_value": 0.0,
|
||||
"cgst": 0.0,
|
||||
"sgst": 0.0,
|
||||
"igst": 0.0,
|
||||
"cess": 0.0,
|
||||
},
|
||||
)
|
||||
group["source_ids"].append(row.get("source_id") or "")
|
||||
group["items"].append(row)
|
||||
for field in ("taxable_value", "cgst", "sgst", "igst", "cess"):
|
||||
group[field] = round(group[field] + float(row.get(field) or 0), 2)
|
||||
|
||||
return sorted(
|
||||
grouped.values(),
|
||||
key=lambda x: (x["invoice_date"], x["invoice_no"], x["eway_bill_no"]),
|
||||
reverse=True,
|
||||
)
|
||||
Reference in New Issue
Block a user