from __future__ import annotations import io import json import logging import os import subprocess import tempfile from typing import Any import requests from requests import HTTPError from .config import BotConfig from .prompts import ( CHAT_SYSTEM_PROMPT, INVOICE_OCR_EXTRACTION_SYSTEM_PROMPT, PAYMENT_EVIDENCE_OCR_EXTRACTION_SYSTEM_PROMPT, ) logger = logging.getLogger(__name__) class ApiError(RuntimeError): def __init__(self, message: str, status_code: int | None = None, payload: dict[str, Any] | None = None) -> None: super().__init__(message) self.status_code = status_code self.payload = payload or {} # ─── TESSERACT OCR ──────────────────────────────────────────────────────────── class TesseractOcrClient: """ Wrapper Tesseract OCR — membaca seluruh teks dari gambar. Membutuhkan binary Tesseract yang sudah diinstall atau ditempatkan manual. Konfigurasi via .env: TESSERACT_BIN = ~/tesseract_local/tesseract (path binary) TESSDATA_DIR = ~/tesseract_local/tessdata (folder data bahasa) TESSERACT_LANG = ind+eng (bahasa OCR) Cara cek ketersediaan di server: which tesseract # cek system-wide ~/tesseract_local/tesseract --version # cek manual binary """ def __init__(self, config: BotConfig) -> None: self.bin = config.tesseract_bin self.tessdata_dir = config.tessdata_dir self.lang = config.tesseract_lang def is_available(self) -> bool: """Cek apakah binary Tesseract dapat ditemukan dan dijalankan.""" return os.path.isfile(self.bin) and os.access(self.bin, os.X_OK) def extract_text(self, image_bytes: bytes) -> str: """ Ekstrak seluruh teks dari gambar menggunakan Tesseract. Preprocessing: konversi grayscale, scale up jika kecil, tingkatkan kontras. Return: string teks hasil OCR. Raise RuntimeError jika gagal. """ if not self.is_available(): raise RuntimeError( f"Tesseract tidak ditemukan di '{self.bin}'. " "Salin binary dari server lama atau install terlebih dahulu. " "Lihat instruksi di README." ) processed_image = self._preprocess(image_bytes) temp_image_path: str | None = None try: # Simpan gambar yang sudah dipreprocess ke file sementara with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as f: processed_image.save(f.name, "PNG") temp_image_path = f.name temp_output_prefix = temp_image_path + "_out" cmd = [ self.bin, temp_image_path, temp_output_prefix, "--tessdata-dir", self.tessdata_dir, "-l", self.lang, "--psm", "6", # Assume a single uniform block of text "--oem", "3", # Default OCR Engine Mode (LSTM) ] logger.info("Tesseract OCR: %s", " ".join(cmd)) result = subprocess.run( cmd, capture_output=True, text=True, timeout=60, ) if result.returncode != 0: logger.warning("Tesseract stderr: %s", result.stderr[:500]) txt_file = temp_output_prefix + ".txt" if not os.path.exists(txt_file): raise RuntimeError("Tesseract tidak menghasilkan output teks.") with open(txt_file, "r", encoding="utf-8") as f: text = f.read().strip() os.unlink(txt_file) if not text: raise RuntimeError("Tesseract menghasilkan teks kosong — foto mungkin buram.") logger.info("Tesseract OK: %d karakter terbaca", len(text)) return text except subprocess.TimeoutExpired as exc: raise RuntimeError("Tesseract timeout (>60 detik).") from exc finally: if temp_image_path and os.path.exists(temp_image_path): try: os.unlink(temp_image_path) except OSError: pass @staticmethod def _preprocess(image_bytes: bytes): """ Preprocessing gambar sebelum OCR: - Konversi ke grayscale (mode L) - Scale up ke min 1200px lebar jika terlalu kecil - Tingkatkan kontras dan ketajaman Membutuhkan Pillow (pip install Pillow). """ try: from PIL import Image, ImageEnhance except ImportError as exc: raise RuntimeError( "Pillow belum terinstall. Jalankan: pip install Pillow" ) from exc image = Image.open(io.BytesIO(image_bytes)).convert("L") w, h = image.size if w < 1200: scale = 1200 / w image = image.resize( (int(w * scale), int(h * scale)), Image.LANCZOS ) image = ImageEnhance.Contrast(image).enhance(2.0) image = ImageEnhance.Sharpness(image).enhance(2.0) return image # ───────────────────────────────────────────────────────────────────────────── class TelegramClient: def __init__(self, config: BotConfig) -> None: self.config = config self.session = requests.Session() self.base_url = f"https://api.telegram.org/bot{config.telegram_bot_token}" self.file_base_url = f"https://api.telegram.org/file/bot{config.telegram_bot_token}" def send_message( self, chat_id: int, text: str, reply_markup: dict[str, Any] | None = None, parse_mode: str | None = None, ) -> dict[str, Any]: payload: dict[str, Any] = { "chat_id": chat_id, "text": text, } if reply_markup: payload["reply_markup"] = reply_markup if parse_mode: payload["parse_mode"] = parse_mode response = self.session.post( f"{self.base_url}/sendMessage", json=payload, timeout=self.config.telegram_timeout_seconds, ) response.raise_for_status() return response.json() def answer_callback_query(self, callback_query_id: str, text: str) -> dict[str, Any]: response = self.session.post( f"{self.base_url}/answerCallbackQuery", json={"callback_query_id": callback_query_id, "text": text}, timeout=self.config.telegram_timeout_seconds, ) response.raise_for_status() return response.json() def get_file_bytes(self, file_id: str) -> bytes: response = self.session.get( f"{self.base_url}/getFile", params={"file_id": file_id}, timeout=self.config.telegram_timeout_seconds, ) response.raise_for_status() payload = response.json() file_path = payload["result"]["file_path"] file_response = self.session.get( f"{self.file_base_url}/{file_path}", timeout=self.config.telegram_timeout_seconds, ) file_response.raise_for_status() return file_response.content class DeepSeekClient: def __init__(self, config: BotConfig) -> None: self.config = config self.session = requests.Session() def analyze_invoice_text(self, ocr_text: str, caption: str | None = None) -> dict[str, Any]: """ Ekstrak data faktur dari teks OCR mentah menggunakan DeepSeek. Tidak membutuhkan vision — hanya teks hasil Tesseract. """ if not self.config.deepseek_api_key: raise RuntimeError("DEEPSEEK_API_KEY belum diisi.") user_prompt = f"Berikut adalah teks OCR dari foto faktur pembelian apotek:\n\n{ocr_text}" if caption: user_prompt += f"\n\nCatatan dari pengirim: {caption}" response = self._call_api( messages=[ {"role": "system", "content": INVOICE_OCR_EXTRACTION_SYSTEM_PROMPT}, {"role": "user", "content": user_prompt}, ], response_format={"type": "json_object"}, temperature=0.1, ) content = response["choices"][0]["message"]["content"] return _load_json_from_text(content) def analyze_payment_evidence_text(self, ocr_text: str, caption: str | None = None) -> dict[str, Any]: if not self.config.deepseek_api_key: raise RuntimeError("DEEPSEEK_API_KEY belum diisi.") user_prompt = f"Berikut adalah teks OCR dari screenshot bukti pembayaran:\n\n{ocr_text}" if caption: user_prompt += f"\n\nCatatan dari pengirim: {caption}" response = self._call_api( messages=[ {"role": "system", "content": PAYMENT_EVIDENCE_OCR_EXTRACTION_SYSTEM_PROMPT}, {"role": "user", "content": user_prompt}, ], response_format={"type": "json_object"}, temperature=0.1, ) content = response["choices"][0]["message"]["content"] return _load_json_from_text(content) def chat( self, user_message: str, history: list[dict[str, str]] | None = None, ) -> str: """ Chat bebas dengan DeepSeek. history: daftar pesan sebelumnya [{"role": "user"|"assistant", "content": "..."}] Return: string jawaban assistant. """ if not self.config.deepseek_api_key: raise RuntimeError("DEEPSEEK_API_KEY belum diisi.") messages: list[dict[str, Any]] = [ {"role": "system", "content": CHAT_SYSTEM_PROMPT}, ] if history: messages.extend(history) messages.append({"role": "user", "content": user_message}) response = self._call_api(messages=messages, temperature=0.7) return response["choices"][0]["message"]["content"] def _call_api( self, messages: list[dict[str, Any]], temperature: float = 0.7, response_format: dict[str, Any] | None = None, ) -> dict[str, Any]: """Helper untuk memanggil DeepSeek Chat Completions API.""" if not self.config.deepseek_api_key: raise RuntimeError("DEEPSEEK_API_KEY belum diisi.") payload: dict[str, Any] = { "model": self.config.deepseek_model, "messages": messages, "temperature": temperature, } if response_format: payload["response_format"] = response_format response = self.session.post( f"{self.config.deepseek_base_url}/chat/completions", headers={ "Authorization": f"Bearer {self.config.deepseek_api_key}", "Content-Type": "application/json", }, json=payload, timeout=self.config.deepseek_timeout_seconds, ) response.raise_for_status() return response.json() class LaravelApiClient: def __init__(self, config: BotConfig) -> None: self.config = config self.session = requests.Session() self.session.headers.update( { "Accept": "application/json", "Authorization": f"Bearer {config.laravel_api_token}", } ) def search_products(self, payload: dict[str, Any]) -> dict[str, Any]: return self._post("/api/bot/products/search", payload) def knowledge_query(self, payload: dict[str, Any]) -> dict[str, Any]: return self._post("/api/bot/products/knowledge-query", payload) def free_chat(self, payload: dict[str, Any]) -> dict[str, Any]: """ Kirim pesan bebas ke engine AI terpusat di Laravel. Menggantikan panggilan langsung ke DeepSeek dari Python. payload: {"query": str, "history": list[dict], "telegram_user_id": str} """ return self._post("/api/bot/chat/free", payload) def get_bot_profile(self, payload: dict[str, Any]) -> dict[str, Any]: return self._get("/api/bot/auth/me", payload) def get_product(self, product_id: int, params: dict[str, Any] | None = None) -> dict[str, Any]: return self._get(f"/api/bot/products/{product_id}", params) def drug_lookup(self, product_id: int, params: dict[str, Any] | None = None) -> dict[str, Any]: """Lookup komposisi obat dari sumber eksternal (Halodoc/Alodokter) untuk produk tertentu.""" return self._get(f"/api/bot/products/{product_id}/drug-lookup", params) def drug_lookup_by_name(self, name: str, params: dict[str, Any] | None = None) -> dict[str, Any]: """Lookup komposisi obat dari sumber eksternal berdasarkan NAMA (tanpa product_id). Digunakan saat produk berasal dari GPOS dan belum ada di DEWA.""" payload: dict[str, Any] = {"name": name} if params: payload.update(params) return self._post("/api/bot/drugs/lookup-by-name", payload) def save_product_composition(self, product_id: int, payload: dict[str, Any], params: dict[str, Any] | None = None) -> dict[str, Any]: """Simpan data komposisi (zat_aktif, nama_generik, komposisi) ke produk DEWA.""" if params: payload = {**payload, **params} return self._post(f"/api/bot/products/{product_id}/composition", payload) def drug_lookup_by_slug(self, slug: str, params: dict[str, Any] | None = None) -> dict[str, Any]: """Ambil detail varian spesifik dari Halodoc berdasarkan slug.""" payload: dict[str, Any] = {"slug": slug} if params: payload.update(params) return self._post("/api/bot/drugs/lookup-by-slug", payload) def save_product_alias(self, payload: dict[str, Any]) -> dict[str, Any]: return self._post("/api/bot/products/aliases", payload) def create_product_candidate(self, payload: dict[str, Any]) -> dict[str, Any]: return self._post("/api/bot/product-candidates", payload) def screening_products(self, payload: dict[str, Any] | None = None) -> dict[str, Any]: """Screening produk dengan harga abnormal (< 50 rupiah) dan stok > 0.""" return self._post("/api/bot/products/screening", payload or {}) def create_defecta(self, payload: dict[str, Any]) -> dict[str, Any]: return self._post("/api/bot/defecta", payload) def sync_product_from_gpos(self, plu: str) -> dict[str, Any]: """Sync satu produk dari GPOS ke DEWA via PLU. Dipakai bot untuk quick sync sebelum defecta.""" return self._post("/api/bot/gpos/sync-product", {"plu": plu}) def delete_defecta(self, defecta_id: int, telegram_user_id: str) -> dict[str, Any]: return self._delete(f"/api/bot/defecta/{defecta_id}", json={"telegram_user_id": telegram_user_id}) def create_purchase_invoice(self, payload: dict[str, Any]) -> dict[str, Any]: return self._post("/api/bot/supplier-invoices", payload) def check_defecta(self, product_id: int, telegram_user_id: str) -> dict[str, Any]: return self._get("/api/bot/defecta/check", {"product_id": product_id, "telegram_user_id": telegram_user_id}) def get_defecta_history(self, product_id: int, telegram_user_id: str) -> dict[str, Any]: return self._get("/api/bot/defecta/history", {"product_id": product_id, "telegram_user_id": telegram_user_id}) def get_open_defecta(self, telegram_user_id: str, limit: int = 50, date: str = "") -> dict[str, Any]: params: dict[str, Any] = {"telegram_user_id": telegram_user_id, "limit": limit} if date: params["date"] = date return self._get("/api/bot/defecta/open", params) def get_ordered_defecta(self, telegram_user_id: str, limit: int = 50) -> dict[str, Any]: return self._get("/api/bot/defecta/ordered", {"telegram_user_id": telegram_user_id, "limit": limit}) def search_suppliers(self, query: str, telegram_user_id: str) -> dict[str, Any]: return self._get("/api/bot/suppliers/search", {"query": query, "telegram_user_id": telegram_user_id}) def start_purchase_order_session(self, payload: dict[str, Any]) -> dict[str, Any]: return self._post("/api/bot/purchase-order-sessions", payload) def add_purchase_order_session_item(self, session_id: int, payload: dict[str, Any]) -> dict[str, Any]: return self._post(f"/api/bot/purchase-order-sessions/{session_id}/items", payload) def cancel_purchase_order_session(self, session_id: int, payload: dict[str, Any]) -> dict[str, Any]: return self._post(f"/api/bot/purchase-order-sessions/{session_id}/cancel", payload) def finalize_purchase_order_session(self, session_id: int, payload: dict[str, Any]) -> dict[str, Any]: return self._post(f"/api/bot/purchase-order-sessions/{session_id}/finalize", payload) def get_purchase_order(self, purchase_order_id: int, telegram_user_id: str) -> dict[str, Any]: return self._get(f"/api/bot/purchase-orders/{purchase_order_id}", {"telegram_user_id": telegram_user_id}) def decide_purchase_order(self, purchase_order_id: int, payload: dict[str, Any]) -> dict[str, Any]: return self._post(f"/api/bot/purchase-orders/{purchase_order_id}/decision", payload) def create_purchase_order(self, payload: dict[str, Any]) -> dict[str, Any]: raise RuntimeError("Gunakan flow purchase-order-sessions untuk membuat PO via bot.") def create_telegram_evidence( self, payload: dict[str, Any], image_bytes: bytes, filename: str = "telegram-evidence.jpg", ) -> dict[str, Any]: return self._post_multipart( "/api/bot/telegram-evidences", payload, file_field="attachment", filename=filename, file_bytes=image_bytes, ) def suggest_supplier_bank_account(self, evidence_id: int, payload: dict[str, Any]) -> dict[str, Any]: return self._post(f"/api/bot/telegram-evidences/{evidence_id}/suggest-bank-account", payload) def create_stock_opname_session(self, payload: dict[str, Any]) -> dict[str, Any]: return self._post("/api/bot/stock-opname/sessions", payload) def get_active_stock_opname_sessions(self, telegram_user_id: str) -> dict[str, Any]: return self._get("/api/bot/stock-opname/sessions/active", {"telegram_user_id": telegram_user_id}) def lock_stock_opname_item(self, session_id: int, payload: dict[str, Any]) -> dict[str, Any]: return self._post(f"/api/bot/stock-opname/sessions/{session_id}/lock", payload) def count_stock_opname_item(self, session_id: int, payload: dict[str, Any]) -> dict[str, Any]: return self._post(f"/api/bot/stock-opname/sessions/{session_id}/count", payload) def get_pending_stock_opname_items(self, session_id: int, telegram_user_id: str, limit: int = 50) -> dict[str, Any]: return self._get( f"/api/bot/stock-opname/sessions/{session_id}/pending", {"telegram_user_id": telegram_user_id, "limit": limit}, ) def get_purchase_history(self, product_id: int, telegram_user_id: str, limit: int = 5) -> dict[str, Any]: return self._get( f"/api/bot/history/products/{product_id}/purchases", {"telegram_user_id": telegram_user_id, "limit": limit}, ) def get_sales_history(self, product_id: int, telegram_user_id: str, limit: int = 5) -> dict[str, Any]: return self._get( f"/api/bot/history/products/{product_id}/sales", {"telegram_user_id": telegram_user_id, "limit": limit}, ) def lookup_purchase_invoice(self, invoice_number: str, telegram_user_id: str) -> dict[str, Any]: return self._get( "/api/bot/history/purchase-invoices/find", {"invoice_number": invoice_number, "telegram_user_id": telegram_user_id}, ) def _post(self, path: str, payload: dict[str, Any]) -> dict[str, Any]: if not self.config.laravel_api_base_url: raise RuntimeError("LARAVEL_API_BASE_URL belum diisi.") response = self.session.post( f"{self.config.laravel_api_base_url}{path}", json=payload, timeout=self.config.laravel_timeout_seconds, verify=self.config.laravel_verify_ssl, ) self._raise_for_status(response) return response.json() def _delete(self, path: str, json: dict[str, Any] | None = None) -> dict[str, Any]: if not self.config.laravel_api_base_url: raise RuntimeError("LARAVEL_API_BASE_URL belum diisi.") kwargs: dict[str, Any] = { "timeout": self.config.laravel_timeout_seconds, "verify": self.config.laravel_verify_ssl, } if json is not None: kwargs["json"] = json response = self.session.delete( f"{self.config.laravel_api_base_url}{path}", **kwargs, ) self._raise_for_status(response) return response.json() def _get(self, path: str, params: dict[str, Any] | None = None) -> dict[str, Any]: if not self.config.laravel_api_base_url: raise RuntimeError("LARAVEL_API_BASE_URL belum diisi.") response = self.session.get( f"{self.config.laravel_api_base_url}{path}", params=params, timeout=self.config.laravel_timeout_seconds, verify=self.config.laravel_verify_ssl, ) self._raise_for_status(response) return response.json() def _post_multipart( self, path: str, payload: dict[str, Any], file_field: str, filename: str, file_bytes: bytes, ) -> dict[str, Any]: if not self.config.laravel_api_base_url: raise RuntimeError("LARAVEL_API_BASE_URL belum diisi.") response = self.session.post( f"{self.config.laravel_api_base_url}{path}", data={key: "" if value is None else str(value) for key, value in payload.items()}, files={file_field: (filename, io.BytesIO(file_bytes), "application/octet-stream")}, timeout=self.config.laravel_timeout_seconds, verify=self.config.laravel_verify_ssl, ) self._raise_for_status(response) return response.json() @staticmethod def _raise_for_status(response: requests.Response) -> None: try: response.raise_for_status() except HTTPError as exc: message = None payload: dict[str, Any] | None = None try: payload = response.json() message = payload.get("message") if not message and isinstance(payload.get("errors"), dict): first_error = next(iter(payload["errors"].values()), []) if first_error: message = first_error[0] except Exception: message = None if message: raise ApiError(message, response.status_code, payload) from exc raise def _load_json_from_text(text: str) -> dict[str, Any]: text = text.strip() if text.startswith("```"): lines = text.splitlines() if len(lines) >= 3: text = "\n".join(lines[1:-1]).strip() return json.loads(text)