Bot & Automation
jorgasisten
/root/hermes-projects/jorgasisten
apps/jorgasisten/services/formatters.py
text
import re
from datetime import date, datetime
from typing import Any
from jorgasisten.schemas import SheetRow, SheetSchema
from jorgasisten.services.text import normalize_key, safe_string, tokenize
HEADER_ALIAS_GROUPS = [
(
{"stok", "stock", "qty", "jumlah", "kuantitas", "sisa"},
{"stok", "stock", "qty", "jumlah", "kuantitas", "sisa"},
),
({"harga", "price", "biaya", "nominal", "tarif"}, {"harga", "price", "biaya", "nominal", "tarif"}),
({"status", "state", "kondisi"}, {"status", "state", "kondisi"}),
({"tanggal", "date", "periode"}, {"tanggal", "date", "periode"}),
(
{"area", "dimana", "mana", "letak", "lokasi", "posisi", "rak", "berada", "lemari"},
{"area", "lokasi", "posisi", "rak", "lemari", "letak", "tempat", "gudang"},
),
({"nama", "customer", "pelanggan", "user"}, {"nama", "customer", "pelanggan", "user"}),
({"produk", "barang", "item"}, {"produk", "barang", "item"}),
({"wa", "whatsapp", "phone", "hp", "telepon"}, {"wa", "whatsapp", "phone", "hp", "telepon"}),
({"invoice", "faktur", "nota"}, {"invoice", "faktur", "nota"}),
]
IDENTITY_HEADER_TOKENS = {"nama", "produk", "barang", "kode", "id", "invoice", "no", "nomor", "wa", "motor"}
EXISTENCE_TOKENS = {"ada", "apakah"}
DISPLAY_TOKENS = {"tampilkan", "lihat", "detail", "semua", "data"}
LOCATION_TOKENS = {"area", "dimana", "mana", "letak", "lokasi", "posisi", "rak", "berada", "lemari", "tempat", "gudang"}
QUANTITY_TOKENS = {"stok", "stock", "qty", "jumlah", "kuantitas", "kuantiti", "sisa"}
CONDITION_TOKENS = {"status", "state", "kondisi"}
PRODUCT_TOKENS = {"produk", "barang", "item"}
PERSON_TOKENS = {"customer", "pelanggan", "user", "nama"}
NAME_TOKENS = {"nama", "produk", "item"}
CODE_TOKENS = {"kode", "id", "invoice", "nomor", "no"}
def format_catalog(catalog: list[SheetSchema]) -> str:
if not catalog:
return "belum ada sheet yang bisa gw baca nih pren."
lines: list[str] = ["nih sheet yang kebaca, pren:"]
for index, schema in enumerate(catalog, start=1):
headers = ", ".join(schema.headers) if schema.headers else "tanpa header"
lines.append(f"{index}. {schema.title} ({schema.row_count} row)")
lines.append(f" kolom: {headers}")
if schema.context_lines:
lines.append(f" konteks: {schema.context_lines[0]}")
if schema.field_options:
option_headers = ", ".join(list(schema.field_options.keys())[:4])
lines.append(f" dropdown: {option_headers}")
return "\n".join(lines)
def format_rows(rows: list[SheetRow], max_rows: int = 8) -> str:
if not rows:
return "belum ketemu datanya di google sheets, pren."
blocks = []
for row in rows[:max_rows]:
visible_items = []
for key, value in row.values.items():
value_text = safe_string(value)
if value_text:
visible_items.append(f"{key}: {value_text}")
if len(visible_items) >= 8:
break
blocks.append(f"sheet: {row.sheet_name}\nrow: {row.row_number}\n" + "\n".join(visible_items))
extra = ""
if len(rows) > max_rows:
extra = f"\n\nmasih ada {len(rows) - max_rows} data lain nih. coba filter-nya diperjelas yaa."
return "\n\n---\n\n".join(blocks) + extra
def build_exact_read_answer(user_message: str, sheet_name: str, rows: list[SheetRow], max_rows: int = 5) -> str:
if not rows:
return "data belum ketemu, pren."
row = rows[0]
selected_headers = refine_requested_headers(
user_message,
select_requested_headers(user_message, list(row.values.keys())),
)
identity_headers = select_identity_headers(list(row.values.keys()))
non_identity_headers = [header for header in selected_headers if header not in identity_headers]
message_tokens = tokenize(user_message)
if message_tokens & EXISTENCE_TOKENS and not non_identity_headers:
return "ada, pren."
if len(rows) > 1:
compact_multi_answer = build_compact_multi_read_answer(
user_message,
rows,
non_identity_headers,
max_rows=max_rows,
)
if compact_multi_answer:
return compact_multi_answer
return "gw nemu beberapa data yang cocok, pren."
if len(non_identity_headers) == 1:
header = non_identity_headers[0]
return concise_field_answer(header, safe_string(row.values.get(header)))
if len(non_identity_headers) > 1:
lines = [
f"- {header}: {safe_string(row.values.get(header))}"
for header in non_identity_headers
if safe_string(row.values.get(header))
]
return "\n".join(lines) if lines else "data ada, tapi nilainya kosong, pren."
if message_tokens & DISPLAY_TOKENS:
visible_headers = [key for key, value in row.values.items() if safe_string(value)][:6]
return "\n".join(f"- {header}: {safe_string(row.values.get(header))}" for header in visible_headers)
identity_headers = select_identity_headers(list(row.values.keys()))
if identity_headers:
identity_header = identity_headers[0]
identity_value = safe_string(row.values.get(identity_header))
if identity_value:
return f"{identity_header}: {identity_value}"
first_value = next((safe_string(value) for value in row.values.values() if safe_string(value)), "")
return first_value or f"data ketemu di sheet {sheet_name}, pren."
def build_grounded_data_answer(user_message: str, sheet_name: str, rows: list[SheetRow], max_rows: int = 5) -> str:
"""Build a deterministic answer using only values read from the selected sheet."""
filtered_rows = filter_rows_by_date_reference(user_message, rows)
if not filtered_rows:
return (
"Data yang Anda tanyakan belum ditemukan di Google Sheets. "
"Silakan periksa kembali nama, kode, atau informasi yang digunakan."
)
requested_headers = refine_requested_headers(
user_message,
select_requested_headers(user_message, list(filtered_rows[0].values.keys())),
)
aggregate_header = select_aggregate_header(user_message, list(filtered_rows[0].values.keys()), requested_headers)
message_tokens = tokenize(user_message)
if is_total_request(message_tokens) and aggregate_header:
values = [parse_numeric_value(safe_string(row.values.get(aggregate_header))) for row in filtered_rows]
numeric_values = [value for value in values if value is not None]
if numeric_values:
return f"Total {aggregate_header} adalah {format_numeric_value(sum(numeric_values))}."
if is_count_request(message_tokens, requested_headers):
return f"Jumlah data yang ditemukan adalah {len(filtered_rows)}."
if is_latest_request(message_tokens):
latest_row = select_latest_row(filtered_rows)
return build_exact_read_answer(user_message, sheet_name, [latest_row], max_rows=max_rows)
return build_exact_read_answer(user_message, sheet_name, filtered_rows, max_rows=max_rows)
def is_total_request(message_tokens: set[str]) -> bool:
return bool(message_tokens & {"total", "jumlahkan", "akumulasi", "akumulasinya"})
def is_count_request(message_tokens: set[str], requested_headers: list[str]) -> bool:
if requested_headers:
return False
phrases = {"jumlah", "banyak", "berapa", "count", "baris", "data"}
return bool(message_tokens & phrases) and bool(message_tokens & {"jumlah", "banyak", "count", "baris"})
def is_latest_request(message_tokens: set[str]) -> bool:
return bool(message_tokens & {"terbaru", "terakhir", "latest"})
def select_aggregate_header(user_message: str, headers: list[str], requested_headers: list[str]) -> str | None:
for header in requested_headers:
if is_numeric_header(header):
return header
preferred_tokens = {"total", "nominal", "nilai", "amount", "harga", "biaya", "qty", "jumlah", "stok", "saldo"}
message_tokens = tokenize(user_message)
candidates = [header for header in headers if is_numeric_header(header)]
if not candidates:
return None
mentioned = [header for header in candidates if tokenize(header) & message_tokens]
if mentioned:
return mentioned[0]
preferred = [header for header in candidates if tokenize(header) & preferred_tokens]
return preferred[0] if preferred else candidates[0]
def is_numeric_header(header: str) -> bool:
numeric_tokens = {
"total",
"nominal",
"nilai",
"amount",
"harga",
"biaya",
"qty",
"jumlah",
"stok",
"saldo",
"kuantitas",
}
return bool(tokenize(header) & numeric_tokens)
def filter_rows_by_date_reference(user_message: str, rows: list[SheetRow]) -> list[SheetRow]:
target_year_month = requested_year_month(user_message)
if target_year_month is None:
return rows
date_headers = [header for header in rows[0].values if tokenize(header) & {"tanggal", "date", "waktu", "periode"}]
if not date_headers:
return rows
matching_rows: list[SheetRow] = []
for row in rows:
for header in date_headers:
parsed_date = parse_sheet_date(safe_string(row.values.get(header)))
if parsed_date and (parsed_date.year, parsed_date.month) == target_year_month:
matching_rows.append(row)
break
return matching_rows
def requested_year_month(user_message: str) -> tuple[int, int] | None:
lowered = user_message.lower()
today = date.today()
if "bulan ini" in lowered:
return today.year, today.month
match = re.search(r"\b(20\d{2})[-/](0?[1-9]|1[0-2])\b", lowered)
if match:
return int(match.group(1)), int(match.group(2))
return None
def parse_sheet_date(value: str) -> date | None:
normalized = value.strip()
for date_format in ("%d/%m/%Y", "%d-%m-%Y", "%Y-%m-%d", "%Y/%m/%d"):
try:
return datetime.strptime(normalized, date_format).date()
except ValueError:
continue
return None
def select_latest_row(rows: list[SheetRow]) -> SheetRow:
date_headers = [header for header in rows[0].values if tokenize(header) & {"tanggal", "date", "waktu", "periode"}]
if not date_headers:
return max(rows, key=lambda row: row.row_number)
def sort_key(row: SheetRow) -> tuple[date, int]:
parsed_dates = [parse_sheet_date(safe_string(row.values.get(header))) for header in date_headers]
latest_date = max((value for value in parsed_dates if value is not None), default=date.min)
return latest_date, row.row_number
return max(rows, key=sort_key)
def format_summary(sheet_name: str, rows: list[SheetRow]) -> str:
if not rows:
return f"belum ada data yang bisa gw ringkas dari sheet {sheet_name}, pren."
filled_columns: dict[str, int] = {}
for row in rows:
for key, value in row.values.items():
if safe_string(value):
filled_columns[key] = filled_columns.get(key, 0) + 1
top_columns = sorted(filled_columns.items(), key=lambda item: item[1], reverse=True)[:6]
lines = [f"nih rekap sheet {sheet_name}, pren", f"total data kebaca: {len(rows)}"]
if top_columns:
lines.append("kolom yang paling rame isinya:")
lines.extend(f"- {key}: {count} data" for key, count in top_columns)
latest = rows[-3:]
if latest:
lines.append("contoh data terakhir:")
for row in latest:
sample = ", ".join(
f"{key}={safe_string(value)}" for key, value in list(row.values.items())[:4] if safe_string(value)
)
lines.append(f"- Row {row.row_number}: {sample or '-'}")
return "\n".join(lines)
def select_requested_headers(user_message: str, headers: list[str]) -> list[str]:
normalized_message = normalize_key(user_message)
message_tokens = tokenize(user_message)
direct_selected: list[str] = []
alias_selected: list[str] = []
identity_headers = set(select_identity_headers(headers))
for header in headers:
normalized_header = normalize_key(header)
header_tokens = tokenize(header)
if normalized_header and normalized_header in normalized_message:
direct_selected.append(header)
continue
if header_tokens and header_tokens.issubset(message_tokens):
direct_selected.append(header)
continue
if header_matches_alias(message_tokens, header_tokens):
alias_selected.append(header)
direct_non_identity = [header for header in direct_selected if header not in identity_headers]
if direct_non_identity:
return direct_selected
alias_non_identity = [header for header in alias_selected if header not in identity_headers]
if alias_non_identity:
return unique_headers([*direct_selected, *alias_non_identity])
return direct_selected or alias_selected
def refine_requested_headers(user_message: str, headers: list[str]) -> list[str]:
if len(headers) <= 1:
return headers
message_tokens = tokenize(user_message)
preference_rules = [
({"sisa", "akhir"}, {"akhir", "sisa"}),
({"stok", "stock"}, {"akhir", "sisa"}),
({"awal"}, {"awal"}),
({"masuk"}, {"masuk"}),
({"keluar"}, {"keluar"}),
({"area"}, {"area"}),
({"lemari"}, {"lemari"}),
({"rak"}, {"rak"}),
]
for trigger_tokens, preferred_header_tokens in preference_rules:
if not (message_tokens & trigger_tokens):
continue
narrowed = [
header
for header in headers
if tokenize(header) & preferred_header_tokens
]
if narrowed:
return narrowed
return headers
def select_identity_headers(headers: list[str]) -> list[str]:
selected: list[str] = []
for header in headers:
if tokenize(header) & IDENTITY_HEADER_TOKENS:
selected.append(header)
return selected[:3]
def header_matches_alias(message_tokens: set[str], header_tokens: set[str]) -> bool:
for alias_tokens, target_tokens in HEADER_ALIAS_GROUPS:
if not (message_tokens & alias_tokens):
continue
if header_tokens & target_tokens:
return True
return False
def unique_headers(headers: list[str]) -> list[str]:
seen: set[str] = set()
result: list[str] = []
for header in headers:
normalized = normalize_key(header)
if not normalized or normalized in seen:
continue
seen.add(normalized)
result.append(header)
return result
def concise_field_answer(header: str, value: str) -> str:
if not value:
return "datanya ada, tapi nilainya kosong, pren."
header_tokens = tokenize(header)
if header_tokens & {"area", "lokasi", "posisi", "rak", "lemari", "letak", "tempat", "gudang"}:
return f"lokasinya di {value}, pren."
if header_tokens & {"stok", "stock", "qty", "jumlah", "kuantitas", "sisa"}:
return f"stoknya {value}, pren."
if header_tokens & {"harga", "price", "biaya", "nominal", "tarif"}:
return f"harganya {value}, pren."
if header_tokens & {"status", "state", "kondisi"}:
return f"statusnya {value}, pren."
if header_tokens & {"tanggal", "date", "periode"}:
return f"tanggalnya {value}, pren."
if header_tokens & {"asal", "daerah", "kota", "origin"}:
return f"asalnya dari {value}, pren."
if header_tokens & {"nama", "customer", "pelanggan", "user"}:
return f"namanya {value}, pren."
return value
def concise_field_label(header: str) -> str:
header_tokens = tokenize(header)
if header_tokens & {"area", "lokasi", "posisi", "rak", "lemari", "letak", "tempat", "gudang"}:
return "lokasinya"
if header_tokens & {"stok", "stock", "qty", "jumlah", "kuantitas", "sisa"}:
return "stoknya"
if header_tokens & {"harga", "price", "biaya", "nominal", "tarif"}:
return "harganya"
if header_tokens & {"status", "state", "kondisi"}:
return "statusnya"
if header_tokens & {"tanggal", "date", "periode"}:
return "tanggalnya"
if header_tokens & {"asal", "daerah", "kota", "origin"}:
return "asalnya"
if header_tokens & {"nama", "customer", "pelanggan", "user"}:
return "namanya"
return header.lower()
def build_compact_multi_read_answer(
user_message: str,
rows: list[SheetRow],
non_identity_headers: list[str],
*,
max_rows: int,
) -> str:
if not rows:
return ""
grouped_location_answer = build_grouped_location_answer(user_message, rows, max_rows=max_rows)
if grouped_location_answer:
return grouped_location_answer
identity_label = build_shared_identity_label(rows)
intro = build_multi_row_intro(rows[0].sheet_name, identity_label)
if len(non_identity_headers) == 1:
header = non_identity_headers[0]
values = [safe_string(row.values.get(header)) for row in rows if safe_string(row.values.get(header))]
unique_values = unique_values_in_order(values)
if len(unique_values) == 1:
return concise_field_answer(header, unique_values[0])
if len(unique_values) > 1:
variant_lines = build_distinct_identity_value_lines(rows, header, max_rows=max_rows)
if len(variant_lines) > 1:
return (
f"gw nemu {len(variant_lines)} data yang mirip. pilih nama atau varian yang dimaksud ya:\n"
+ "\n".join(variant_lines)
)
joined_values = ", ".join(unique_values[:max_rows])
field_label = concise_field_label(header)
if len(unique_values) == 2:
return f"gw nemu 2 data yang cocok. {field_label}: {joined_values}."
return f"gw nemu beberapa data yang cocok. {field_label}: {joined_values}."
body = "\n".join(f"- {value}" for value in unique_values[:max_rows])
return f"{intro}\n{body}" if intro else body
preview_lines: list[str] = []
for row in rows[:max_rows]:
identity_text = row_identity_text(row)
if non_identity_headers:
details = ", ".join(
f"{header}: {safe_string(row.values.get(header))}"
for header in non_identity_headers
if safe_string(row.values.get(header))
)
if details:
preview_lines.append(f"- {identity_text} | {details}")
continue
preview_lines.append(f"- {identity_text}")
body = "\n".join(preview_lines)
return f"{intro}\n{body}" if intro else body
def build_distinct_identity_value_lines(rows: list[SheetRow], header: str, *, max_rows: int) -> list[str]:
lines: list[str] = []
seen: set[tuple[str, str]] = set()
identities: set[str] = set()
for row in rows:
identity = row_identity_text(row)
value = safe_string(row.values.get(header))
if not identity or not value:
continue
normalized_identity = normalize_key(identity)
fingerprint = (normalize_key(identity), normalize_key(value))
if fingerprint in seen:
continue
seen.add(fingerprint)
identities.add(normalized_identity)
lines.append(f"- {identity}: {header} {value}")
if len(lines) >= max_rows:
break
return lines if len(identities) > 1 else []
def row_identity_text(row: SheetRow) -> str:
for header in select_identity_headers(list(row.values.keys())):
value = safe_string(row.values.get(header))
if value:
return value
for key, value in row.values.items():
value_text = safe_string(value)
if value_text:
return f"{key}: {value_text}"
return f"row {row.row_number}"
def build_grouped_location_answer(user_message: str, rows: list[SheetRow], *, max_rows: int) -> str:
if not rows:
return ""
message_tokens = tokenize(user_message)
headers = collect_headers(rows)
area_header = first_header_by_tokens(headers, {"area", "lokasi", "tempat", "gudang"})
storage_header = first_header_by_tokens(headers, {"lemari", "rak", "slot", "bin", "shelf", "posisi"})
quantity_header = first_header_by_tokens(headers, QUANTITY_TOKENS)
condition_header = first_header_by_tokens(headers, CONDITION_TOKENS)
is_location_question = bool(message_tokens & LOCATION_TOKENS)
if not is_location_question:
return ""
if not area_header and not storage_header:
return ""
identity_label = build_shared_identity_label(rows)
intro = build_multi_row_intro(
rows[0].sheet_name,
identity_label,
purpose="detail lokasi",
)
if len(rows) == 1:
row = rows[0]
area_value = safe_string(row.values.get(area_header)) if area_header else ""
storage_value = safe_string(row.values.get(storage_header)) if storage_header else ""
location_parts = [value for value in [area_value, storage_value] if value]
if location_parts:
return f"{intro}\n" + "\n".join(location_parts) if intro else concise_field_answer(
storage_header or area_header or "Lokasi",
", ".join(location_parts),
)
return ""
if area_header:
grouped_lines = build_area_group_lines(
rows[:max_rows],
area_header=area_header,
storage_header=storage_header,
quantity_header=quantity_header,
condition_header=condition_header,
)
if grouped_lines:
total_line = build_quantity_total_line(rows, quantity_header)
parts = [intro, *grouped_lines]
if total_line:
parts.append(total_line)
return "\n".join(part for part in parts if part)
if storage_header:
detail_lines = [
build_storage_detail_line(
row,
storage_header=storage_header,
quantity_header=quantity_header,
condition_header=condition_header,
)
for row in rows[:max_rows]
]
detail_lines = [line for line in detail_lines if line]
if detail_lines:
total_line = build_quantity_total_line(rows, quantity_header)
parts = [intro, *detail_lines]
if total_line:
parts.append(total_line)
return "\n".join(part for part in parts if part)
return ""
def build_area_group_lines(
rows: list[SheetRow],
*,
area_header: str,
storage_header: str | None,
quantity_header: str | None,
condition_header: str | None,
) -> list[str]:
groups: list[tuple[str, list[str]]] = []
group_map: dict[str, int] = {}
detail_map: dict[tuple[str, str, str], int] = {}
aggregated_rows: dict[tuple[str, str, str], SheetRow] = {}
for row in rows:
area_value = safe_string(row.values.get(area_header))
if not area_value:
area_value = "Lokasi lain"
area_key = normalize_key(area_value) or area_value
if area_key not in group_map:
group_map[area_key] = len(groups)
groups.append((area_value, []))
storage_value = safe_string(row.values.get(storage_header)) if storage_header else ""
condition_value = safe_string(row.values.get(condition_header)) if condition_header else ""
detail_key = (area_key, normalize_key(storage_value), normalize_key(condition_value))
existing = aggregated_rows.get(detail_key)
if existing is None:
aggregated_rows[detail_key] = row.model_copy(deep=True)
detail_map[detail_key] = len(aggregated_rows)
continue
if quantity_header:
existing_quantity = parse_numeric_value(safe_string(existing.values.get(quantity_header)))
added_quantity = parse_numeric_value(safe_string(row.values.get(quantity_header)))
if existing_quantity is not None and added_quantity is not None:
existing.values[quantity_header] = format_numeric_value(existing_quantity + added_quantity)
ordered_details = sorted(detail_map, key=detail_map.get)
for detail_key in ordered_details:
area_key = detail_key[0]
index = group_map[area_key]
_, lines = groups[index]
detail_line = build_storage_detail_line(
aggregated_rows[detail_key],
storage_header=storage_header,
quantity_header=quantity_header,
condition_header=condition_header,
)
if detail_line:
lines.append(detail_line)
output: list[str] = []
for area_value, detail_lines in groups:
output.append(f"Area: {area_value}")
if detail_lines:
output.extend(detail_lines)
return output
def build_storage_detail_line(
row: SheetRow,
*,
storage_header: str | None,
quantity_header: str | None,
condition_header: str | None,
) -> str:
storage_value = safe_string(row.values.get(storage_header)) if storage_header else ""
quantity_value = safe_string(row.values.get(quantity_header)) if quantity_header else ""
condition_value = safe_string(row.values.get(condition_header)) if condition_header else ""
if storage_header and storage_value:
line = f"{storage_header} {storage_value}"
else:
fallback_header = quantity_header or condition_header
fallback_value = quantity_value or condition_value
if not fallback_header or not fallback_value:
return ""
line = f"{fallback_header}: {fallback_value}"
if quantity_header and quantity_value:
line = f"{line}: {quantity_header} {quantity_value}"
if condition_header and condition_value:
line = f"{line} ({condition_header}: {condition_value})"
return line
def build_quantity_total_line(rows: list[SheetRow], quantity_header: str | None) -> str:
if not quantity_header:
return ""
total = 0.0
numeric_found = False
for row in rows:
value = safe_string(row.values.get(quantity_header))
parsed_value = parse_numeric_value(value)
if parsed_value is None:
continue
numeric_found = True
total += parsed_value
if not numeric_found:
return ""
quantity_label = "stok" if tokenize(quantity_header) & QUANTITY_TOKENS else quantity_header.lower()
return f"Total keseluruhan {quantity_label} yang tercatat adalah {format_numeric_value(total)}."
def build_shared_identity_label(rows: list[SheetRow]) -> str:
common_pairs = common_identity_pairs(rows)
if not common_pairs:
return ""
name_pair = next(
(
pair
for pair in common_pairs
if (tokenize(pair[0]) & NAME_TOKENS)
or ((tokenize(pair[0]) & {"barang"}) and not (tokenize(pair[0]) & CODE_TOKENS))
),
None,
)
code_pair = next((pair for pair in common_pairs if tokenize(pair[0]) & CODE_TOKENS), None)
if name_pair:
header, value = name_pair
header_tokens = tokenize(header)
if header_tokens & PRODUCT_TOKENS:
label = f"barang {value}"
elif header_tokens & PERSON_TOKENS:
label = value
else:
label = f"{header}: {value}"
if code_pair and normalize_key(code_pair[1]) != normalize_key(value):
label = f"{label} ({code_pair[0]}: {code_pair[1]})"
return label
if code_pair:
return f"{code_pair[0]}: {code_pair[1]}"
header, value = common_pairs[0]
return f"{header}: {value}"
def common_identity_pairs(rows: list[SheetRow]) -> list[tuple[str, str]]:
if not rows:
return []
headers = collect_headers(rows)
common: list[tuple[str, str]] = []
for header in select_identity_headers(headers):
values = [safe_string(row.values.get(header)) for row in rows]
if any(not value for value in values):
continue
normalized_values = {normalize_key(value) for value in values}
if len(normalized_values) == 1:
common.append((header, values[0]))
return common
def build_multi_row_intro(sheet_name: str, identity_label: str, *, purpose: str = "detail data") -> str:
if identity_label:
return f"Berdasarkan data pada tabel {sheet_name}, berikut {purpose} untuk {identity_label}:"
return f"Berdasarkan data pada tabel {sheet_name}, berikut {purpose}:"
def collect_headers(rows: list[SheetRow]) -> list[str]:
headers: list[str] = []
seen: set[str] = set()
for row in rows:
for header in row.values.keys():
normalized = normalize_key(header)
if not normalized or normalized in seen:
continue
seen.add(normalized)
headers.append(header)
return headers
def first_header_by_tokens(headers: list[str], target_tokens: set[str]) -> str | None:
for header in headers:
if tokenize(header) & target_tokens:
return header
return None
def parse_numeric_value(value: str) -> float | None:
if not value:
return None
cleaned = re.sub(r"[^0-9,.-]", "", value)
if not cleaned or cleaned in {"-", ".", ","}:
return None
if "," in cleaned and "." in cleaned:
if cleaned.rfind(",") > cleaned.rfind("."):
cleaned = cleaned.replace(".", "").replace(",", ".")
else:
cleaned = cleaned.replace(",", "")
elif cleaned.count(",") > 1:
cleaned = cleaned.replace(",", "")
elif cleaned.count(".") > 1:
cleaned = cleaned.replace(".", "")
elif "," in cleaned:
parts = cleaned.split(",")
cleaned = cleaned.replace(",", ".") if len(parts[-1]) <= 2 else cleaned.replace(",", "")
elif "." in cleaned:
parts = cleaned.split(".")
cleaned = cleaned if len(parts[-1]) <= 2 else cleaned.replace(".", "")
try:
return float(cleaned)
except ValueError:
return None
def format_numeric_value(value: float) -> str:
if value.is_integer():
return str(int(value))
return f"{value:.2f}".rstrip("0").rstrip(".")
def unique_values_in_order(values: list[str]) -> list[str]:
seen: set[str] = set()
result: list[str] = []
for value in values:
normalized = normalize_key(value)
if not normalized or normalized in seen:
continue
seen.add(normalized)
result.append(value)
return result
def format_operation_error(error: Exception) -> str:
error_text = safe_string(error).lower()
if "quota exceeded" in error_text or "rate_limit_exceeded" in error_text or "httperror 429" in error_text:
return (
"google sheets lagi padat kena limit sementara, pren. "
"tunggu bentar lalu coba lagi yaa."
)
if "exceeds grid limits" in error_text:
return (
"sheet tujuan sempat mentok batas row, pren. "
"gw sudah siapin flow auto-tambah row, jadi coba kirim lagi yaa."
)
if "permission" in error_text or "the caller does not have permission" in error_text:
return "akses ke sheet ini ketolak, pren. cek share service account sama izin editnya dulu."
if "requested entity was not found" in error_text or "unable to parse range" in error_text:
return "sheet atau range yang dituju gak ketemu, pren. cek nama sheet dan spreadsheet aktifnya yaa."
return (
f"aduh pren, prosesnya belum berhasil: {error.__class__.__name__}. "
"coba cek konfigurasi atau kirim instruksinya lebih jelas yaa."
)
def compact_catalog_for_prompt(catalog: list[SheetSchema]) -> list[dict[str, Any]]:
return [
{
"title": schema.title,
"headers": schema.headers,
"sample_rows": schema.sample_rows,
"context_lines": schema.context_lines,
"field_options": schema.field_options,
"field_validation_messages": schema.field_validation_messages,
"header_row_number": schema.header_row_number,
"row_count": schema.row_count,
}
for schema in catalog
]