Bot & Automation
jorgasisten
/root/hermes-projects/jorgasisten
apps/jorgasisten/services/sheet_router.py
text
import re
from typing import Any
from jorgasisten.schemas import AssistantDecision, Intent, Operation, SheetSchema
from jorgasisten.services.text import compact_search_token_sequence, normalize_key, safe_string, tokenize
WRITE_KEYWORDS = {"catat", "input", "tambah", "simpan", "masukkan", "buat data", "tambahkan"}
UPDATE_KEYWORDS = {"update", "ubah", "ganti", "edit", "perbarui"}
DELETE_KEYWORDS = {"hapus", "delete", "hilangkan"}
SUMMARY_KEYWORDS = {"ringkasan", "rekap", "laporan", "summary", "analisis", "insight", "tren", "bandingkan"}
QUESTION_KEYWORDS = {"apa", "berapa", "ada", "cari", "tampilkan", "lihat", "cek", "siapa", "kapan"}
GREETING_KEYWORDS = {"halo", "haloo", "hai", "hi", "pagi", "siang", "sore", "malam", "makasih", "thanks"}
DATA_HINT_KEYWORDS = {
"asal",
"area",
"barang",
"catatan",
"customer",
"data",
"garansi",
"gudang",
"harga",
"klaim",
"invoice",
"item",
"kategori",
"kode",
"lemari",
"lokasi",
"masa",
"umur",
"sheet",
"spreadsheet",
"booking",
"pricelist",
"produk",
"pelanggan",
"qty",
"rak",
"service",
"status",
"stok",
"stock",
"supplier",
"transaksi",
"penjualan",
"pembelian",
}
READ_FILTER_STOPWORDS = QUESTION_KEYWORDS | DATA_HINT_KEYWORDS | {
"apakah",
"atas",
"berada",
"dimana",
"dong",
"harga",
"id",
"kondisi",
"kuantiti",
"kuantitas",
"lokasinya",
"lemari",
"letak",
"lokasi",
"mana",
"mama",
"ya",
"yaa",
"nih",
"info",
"invoice",
"kode",
"motor",
"nama",
"no",
"nomor",
"pelanggan",
"produk",
"qty",
"sekarang",
"status",
"statusnya",
"stok",
"stoknya",
"suppliernya",
"tanggal",
"tanggalnya",
"tentang",
"untuk",
"wa",
"yang",
"di",
"ke",
}
READ_QUESTION_TERMS = {
"ada",
"apakah",
"berapa",
"harga",
"status",
"stok",
"kapan",
"siapa",
"lokasi",
"dimana",
"mana",
"letak",
"berada",
}
WEAK_FILTER_VALUE_TOKENS = {
"barang",
"data",
"ini",
"item",
"itu",
"produk",
}
DATA_FALLBACK_TOKENS = DATA_HINT_KEYWORDS | {
"alamat",
"biaya",
"daftar",
"faktur",
"jumlah",
"kuantiti",
"kuantitas",
"letak",
"mana",
"nominal",
"nota",
"posisi",
"price",
"riwayat",
"siapa",
"stoknya",
"suppliernya",
"tempat",
}
def score_schema(message: str, schema: SheetSchema) -> int:
message_tokens = tokenize(message)
if not message_tokens:
return 0
schema_text_parts = [schema.title, *schema.headers]
for sample in schema.sample_rows:
schema_text_parts.extend(safe_string(value) for value in sample.values())
score = 0
title_tokens = tokenize(schema.title)
header_tokens = set().union(*(tokenize(header) for header in schema.headers)) if schema.headers else set()
sample_tokens = set().union(*(tokenize(part) for part in schema_text_parts)) if schema_text_parts else set()
score += 5 * len(message_tokens & title_tokens)
score += 3 * len(message_tokens & header_tokens)
score += len(message_tokens & sample_tokens)
normalized_message = normalize_key(message)
if normalize_key(schema.title) and normalize_key(schema.title) in normalized_message:
score += 10
for header in schema.headers:
normalized_header = normalize_key(header)
if normalized_header and normalized_header in normalized_message:
score += 4
return score
def select_best_schema(message: str, catalog: list[SheetSchema]) -> tuple[SheetSchema | None, int]:
if not catalog:
return None, 0
ranked = sorted(
((schema, score_schema(message, schema)) for schema in catalog),
key=lambda item: item[1],
reverse=True,
)
best_schema, best_score = ranked[0]
if best_score <= 0 and len(catalog) == 1:
return best_schema, 45
confidence = min(95, 40 + best_score * 5)
return best_schema, confidence
def parse_key_value_data(message: str, headers: list[str]) -> dict[str, Any]:
data: dict[str, Any] = {}
normalized_headers = {normalize_key(header): header for header in headers}
for raw_line in message.replace(";", "\n").splitlines():
if ":" not in raw_line and "=" not in raw_line:
continue
separator = ":" if ":" in raw_line else "="
key, value = [part.strip() for part in raw_line.split(separator, 1)]
normalized_key = normalize_key(key)
header = normalized_headers.get(normalized_key)
if header and value:
data[header] = value
return data
def classify_operation(message: str) -> tuple[Intent, Operation]:
lowered = message.lower()
if lowered.strip() in GREETING_KEYWORDS:
return "chat_umum", "chat"
if any(keyword in lowered for keyword in DELETE_KEYWORDS):
return "hapus_data", "delete"
if any(keyword in lowered for keyword in UPDATE_KEYWORDS):
return "update_data", "update"
if any(keyword in lowered for keyword in SUMMARY_KEYWORDS):
return "ringkasan_data", "summarize"
if any(keyword in lowered for keyword in WRITE_KEYWORDS):
return "input_data", "create"
if any(keyword in lowered for keyword in QUESTION_KEYWORDS) or message.strip().endswith("?"):
return "tanya_data", "read"
return "chat_umum", "chat"
def structure_decision_from_message(message: str) -> AssistantDecision | None:
lowered = message.lower().strip()
headers = extract_headers(message)
create_match = re.search(
r"(?:buat|bikin|tambah)\s+(?:sheet|tabel|tab)(?:\s+baru)?\s+(.+?)(?:\s+dengan|\s+kolom|$)",
message,
flags=re.IGNORECASE,
)
if create_match:
return AssistantDecision(
intent="struktur_sheet",
sheet_name=clean_sheet_name(create_match.group(1)),
operation="create_sheet",
confidence=92 if headers else 65,
missing_fields=[] if headers else ["kolom/header"],
data={"headers": headers} if headers else {},
reason="Fallback parser membaca instruksi buat sheet/tabel baru.",
user_reply="siap pren, gw siapin sheet barunya yaa.",
)
update_headers_match = re.search(
r"(?:ubah|ganti|update)\s+(?:struktur\s+)?(?:kolom|header)\s+(?:sheet|tabel|tab)?\s*(.+?)\s+(?:jadi|menjadi|dengan)\s+(.+)",
message,
flags=re.IGNORECASE,
)
if update_headers_match:
parsed_headers = split_headers(update_headers_match.group(2))
return AssistantDecision(
intent="struktur_sheet",
sheet_name=clean_sheet_name(update_headers_match.group(1)),
operation="update_headers",
confidence=92 if parsed_headers else 65,
missing_fields=[] if parsed_headers else ["kolom/header baru"],
data={"headers": parsed_headers} if parsed_headers else {},
reason="Fallback parser membaca instruksi ganti header sheet.",
user_reply="oke pren, header sheet-nya gw siapin buat diganti yaa.",
)
rename_match = re.search(
r"(?:rename|ganti\s+nama)\s+(?:sheet|tabel|tab)\s+(.+?)\s+(?:jadi|menjadi|ke)\s+(.+)",
message,
flags=re.IGNORECASE,
)
if rename_match:
return AssistantDecision(
intent="struktur_sheet",
sheet_name=clean_sheet_name(rename_match.group(1)),
operation="rename_sheet",
confidence=92,
data={"new_sheet_name": clean_sheet_name(rename_match.group(2))},
reason="Fallback parser membaca instruksi rename sheet.",
user_reply="siap pren, nama sheet-nya gw ganti yaa.",
)
delete_match = re.search(
r"(?:hapus|delete)\s+(?:sheet|tabel|tab)(?:\s+penuh)?\s+(.+)",
message,
flags=re.IGNORECASE,
)
if delete_match and ("sheet" in lowered or "tabel" in lowered or "tab" in lowered):
return AssistantDecision(
intent="struktur_sheet",
sheet_name=clean_sheet_name(delete_match.group(1)),
operation="delete_sheet",
confidence=92,
data={"confirm_delete": True},
reason="Fallback parser membaca instruksi hapus sheet penuh.",
user_reply="oke pren, gw hapus sheet itu kalau targetnya valid yaa.",
)
return None
def extract_headers(message: str) -> list[str]:
match = re.search(r"(?:dengan\s+)?(?:kolom|header)\s*:?\s+(.+)$", message, flags=re.IGNORECASE)
if not match:
return []
return split_headers(match.group(1))
def split_headers(raw_headers: str) -> list[str]:
cleaned = re.sub(r"\s+(?:yaa|ya|dong|dongss)$", "", raw_headers.strip(), flags=re.IGNORECASE)
parts = re.split(r"\s*,\s*|\s*\|\s*", cleaned)
headers: list[str] = []
seen: set[str] = set()
for part in parts:
header = clean_sheet_name(part)
if not header:
continue
normalized = normalize_key(header)
if normalized in seen:
continue
seen.add(normalized)
headers.append(header)
return headers
def clean_sheet_name(value: str) -> str:
return re.sub(r"\s+", " ", value.strip(" .,:;\"'")).strip()
def read_filters_from_message(message: str) -> dict[str, Any]:
search_terms = [
token
for token in compact_search_token_sequence(message)
if token not in READ_FILTER_STOPWORDS and (len(token) >= 2 or token.isdigit())
]
if not search_terms:
return {}
return {"query": " ".join(search_terms)}
def ordered_tokens(message: str) -> list[str]:
return [token.lower() for token in re.findall(r"[a-z0-9]+", message, flags=re.IGNORECASE)]
def should_prefer_sheet_read(message: str, catalog: list[SheetSchema]) -> bool:
return is_likely_data_question(message, catalog)
def is_likely_data_question(message: str, catalog: list[SheetSchema]) -> bool:
message_tokens = tokenize(message)
if not message_tokens:
return False
if message_tokens & DATA_FALLBACK_TOKENS:
return True
if header_overlap_hint(message_tokens, catalog) >= 2:
return True
schema, confidence = select_best_schema(message, catalog)
if schema is None:
return False
if header_overlap_hint(message_tokens, [schema]) >= 1 and (
message.strip().endswith("?") or message_tokens & QUESTION_KEYWORDS
):
return True
if confidence >= 75:
return True
return confidence >= 60 and message.strip().endswith("?")
def header_overlap_hint(message_tokens: set[str], catalog: list[SheetSchema]) -> int:
best_overlap = 0
for schema in catalog:
header_tokens = set().union(*(tokenize(header) for header in schema.headers)) if schema.headers else set()
overlap = len(message_tokens & header_tokens)
if overlap > best_overlap:
best_overlap = overlap
return best_overlap
def enrich_read_decision(message: str, decision: AssistantDecision, catalog: list[SheetSchema]) -> AssistantDecision:
if decision.operation != "read":
return decision
schema = next((item for item in catalog if item.title == decision.sheet_name), None)
if schema is None:
schema, _ = select_best_schema(message, catalog)
if schema is None:
return decision
filters = build_read_filters(message, schema)
if filters:
return decision.model_copy(update={"sheet_name": schema.title, "filters": filters})
if is_likely_data_question(message, catalog):
return decision.model_copy(
update={
"sheet_name": schema.title,
"operation": "ask_clarification",
"confidence": min(decision.confidence, 55),
"user_reply": (
"pertanyaannya masih belum cukup jelas, pren. "
"sebutin targetnya lebih spesifik, misalnya nama, kode, invoice, atau data yang mau dicek."
),
}
)
return decision
def build_read_filters(message: str, schema: SheetSchema) -> dict[str, Any]:
headers = schema.headers
pattern_filters = extract_pattern_filters(message, headers)
if pattern_filters:
return pattern_filters
return read_filters_from_message(message)
def extract_pattern_filters(message: str, headers: list[str]) -> dict[str, Any]:
patterns = [
(
r"atas nama\s+(.+?)(?:[?.,]|$)",
find_header_by_alias_priority(headers, ["nama", "customer", "pelanggan"]),
),
(
r"(?:invoice|inv)\s+([a-z0-9._/-]+)",
find_header_by_alias_priority(headers, ["invoice", "faktur", "nota", "kode", "id"]),
),
(
r"(?:kode barang|kode|sku|id)\s+([a-z0-9._/-]+)",
find_header_by_alias_priority(headers, ["kode", "sku", "id", "invoice"]),
),
(
r"(?:area|lokasi|letak|rak|lemari|posisi)\s+mana\s+(.+?)(?:[?.,]|$)",
find_header_by_alias_priority(headers, ["nama barang", "nama produk", "produk", "barang", "item", "nama"]),
),
(
r"(?:produk|barang|item)\s+(.+?)(?:\s+(?:ada|berapa|status|stok|harga|kapan|lokasi|dimana|mana|letak|berada|di)\b|[?.,]|$)",
find_header_by_alias_priority(headers, ["nama barang", "nama produk", "produk", "barang", "item", "nama"]),
),
]
for pattern, header in patterns:
if not header:
continue
match = re.search(pattern, message, flags=re.IGNORECASE)
if not match:
continue
value = clean_filter_value(match.group(1))
if value and not is_weak_filter_value(value):
return {header: value}
return {}
def find_header_by_alias(headers: list[str], aliases: set[str]) -> str | None:
for header in headers:
normalized = normalize_key(header)
for alias in aliases:
if normalize_key(alias) and normalize_key(alias) in normalized:
return header
return None
def find_header_by_alias_priority(headers: list[str], aliases: list[str]) -> str | None:
for alias in aliases:
matched = find_header_by_alias(headers, {alias})
if matched:
return matched
return None
def clean_filter_value(value: str) -> str:
cleaned = safe_string(value)
cleaned = re.sub(
r"\b(?:ada|berapa|status|stok|harga|kapan|lokasi|dimana|mana|letak|berada|di)\b.*$",
"",
cleaned,
flags=re.IGNORECASE,
).strip(" -,:;?")
if tokenize(cleaned) & READ_QUESTION_TERMS:
cleaned = " ".join(token for token in ordered_tokens(cleaned) if token not in READ_QUESTION_TERMS)
return cleaned.strip()
def is_weak_filter_value(value: str) -> bool:
tokens = ordered_tokens(value)
return bool(tokens) and set(tokens).issubset(WEAK_FILTER_VALUE_TOKENS)
def fallback_decision(message: str, catalog: list[SheetSchema]) -> AssistantDecision:
structure_decision = structure_decision_from_message(message)
if structure_decision is not None:
return structure_decision
schema, confidence = select_best_schema(message, catalog)
intent, operation = classify_operation(message)
sheet_name = schema.title if schema else ""
data = parse_key_value_data(message, schema.headers if schema else [])
if operation == "create" and not data:
confidence = min(confidence, 55)
likely_data_question = is_likely_data_question(message, catalog)
if operation == "chat" and likely_data_question:
intent = "tanya_data"
operation = "read"
confidence = max(confidence, 65)
sheet_name = schema.title if schema else ""
if operation == "read" and confidence < 60 and not likely_data_question:
intent = "chat_umum"
operation = "chat"
if operation == "chat":
sheet_name = ""
confidence = max(confidence, 70)
if operation == "ask_clarification":
confidence = min(confidence, 45)
missing_fields = []
filters: dict[str, Any] = {}
user_reply = "bisaa pren, tanya aja. gw bantu jawab sebisanya yaa."
if operation == "create" and not data:
missing_fields.append("data dengan format kolom yang sesuai")
user_reply = "data yang mau dicatat belum jelas, bestii. kirim formatnya kayak Nama Kolom: nilai yaa."
elif operation == "read":
filters = read_filters_from_message(message)
user_reply = "gw cekin data yang paling nyambung dari google sheets yaa pren."
elif operation == "summarize":
user_reply = "gw bikinin rekap dari data google sheets yang relevan yaa pren."
return AssistantDecision(
intent=intent,
sheet_name=sheet_name,
operation=operation,
confidence=confidence,
missing_fields=missing_fields,
data=data,
filters=filters,
reason="Fallback parser digunakan karena AI provider tidak tersedia atau tidak mengembalikan JSON valid.",
user_reply=user_reply,
)