from __future__ import annotations import os import uuid from datetime import date from typing import Dict, Optional import requests BASE_URL = "https://api.manorivile.lt/client/v2" TIMEOUT = 30 def _post(api_key: str, payload: dict) -> dict: headers = { "ApiKey": api_key, "Content-Type": "application/json", "Accept": "application/json", } response = requests.post( BASE_URL, json=payload, headers=headers, timeout=TIMEOUT, ) if response.status_code != 200: raise RuntimeError(f"Rivile HTTP {response.status_code}: {response.text}") data = response.json() if "errorMessage" in data: raise RuntimeError(f"Rivile API error: {data}") return data def get_n08_client_by_ks_code(ks_code: str) -> Optional[Dict]: api_key = os.getenv("RIVILE_API_KEY", "").strip() if not api_key: raise RuntimeError("Missing RIVILE_API_KEY environment variable.") payload = { "method": "GET_N08_LIST", "params": { "list": "A", "fil": f"n08_kodas_ks='{ks_code}'", }, } resp = _post(api_key, payload) if "N08" not in resp: return None if isinstance(resp["N08"], list): return resp["N08"][0] if resp["N08"] else None return resp["N08"] def _normalize_optional(value: Optional[str]) -> Optional[str]: if value is None: return None cleaned = value.strip() if not cleaned: return None if cleaned.upper() == "ND": return None return cleaned def _person_type_to_n08(person_type: str) -> str: normalized = (person_type or "").strip().lower() return "2" if normalized == "physical" else "1" def upsert_client_n08( client_code: str, name: str, company_code: Optional[str] = None, vat_code: Optional[str] = None, country: str = "LT", city: Optional[str] = None, address: Optional[str] = None, email: Optional[str] = None, phone: Optional[str] = None, operation: str = "I", existing_code: Optional[str] = None, person_type: str = "legal", ) -> str: api_key = os.getenv("RIVILE_API_KEY", "").strip() if not api_key: raise RuntimeError("Missing RIVILE_API_KEY environment variable.") if operation not in {"I", "U"}: raise RuntimeError("Invalid operation for N08 payload.") normalized_company = _normalize_optional(company_code) normalized_vat = _normalize_optional(vat_code) if person_type == "physical": normalized_company = None normalized_vat = None n08_data = { "N08_KODAS": client_code, "N08_KODAS_KS": client_code, "N08_PAV": name, "N08_TIPAS": _person_type_to_n08(person_type), "N08_IM_KODAS": normalized_company or "", "N08_PVM_KODAS": normalized_vat or "", "N08_SALIS": country, "N08_MIESTAS": city or "", "N08_ADRESAS": address or "", "N08_EL_PASTAS": email or "", "N08_MOB_TEL": phone or "", "N08_AKTYVUS": "T", } if operation == "U" and existing_code: n08_data["N08_KODAS"] = existing_code if operation == "U": payload = { "method": "EDIT_N08", "params": { "oper": "U", "user": api_key.split(".", 1)[0], "fld": "N08_KODAS_KS", "val": client_code, }, "data": {"N08": n08_data}, } else: payload = { "method": "EDIT_N08_FULL", "params": {"oper": "I"}, "data": {"N08": n08_data}, } _post(api_key, payload) return client_code def create_contract_n51( client_code: str, title: str, start_date: date, end_date: Optional[date] = None, comment: Optional[str] = None, status_code: Optional[str] = None, contract_type: Optional[str] = None, ) -> str: api_key = os.getenv("RIVILE_API_KEY", "").strip() if not api_key: raise RuntimeError("Missing RIVILE_API_KEY environment variable.") contract_code = f"SUT-{uuid.uuid4().hex[:8].upper()}" n51 = { "N51_KODAS_KT": contract_code, "N51_KODAS_KS": client_code, "N51_PAV": title, "N51_OP_DATA": start_date.isoformat(), "N51_BEG_DATE": start_date.isoformat(), "N51_PASTABOS": comment or "", } if end_date: n51["N51_END_DATE"] = end_date.isoformat() if status_code is not None: n51["N51_VISKAS"] = status_code if contract_type: n51["N51_TIPAS"] = contract_type payload = { "method": "EDIT_N51_FULL", "params": { "oper": "I", "user": api_key.split(".", 1)[0], }, "data": {"N51": n51}, } _post(api_key, payload) return contract_code def create_contract_appendix_n52( contract_code: str, service_code: str, price: float, start_date: date, end_date: Optional[date] = None, quantity: float = 1.0, line_no: int = 1, price_list_code: Optional[str] = None, item_type: str = "service", ) -> None: api_key = os.getenv("RIVILE_API_KEY", "").strip() if not api_key: raise RuntimeError("Missing RIVILE_API_KEY environment variable.") rusis = "2" if item_type == "service" else "1" op_date = date.today().isoformat() n52 = { "N52_EIL_NR": line_no, "N52_KODAS_KT": contract_code, "N52_RUSIS": rusis, "N52_KODAS": service_code, "N52_KIEKIS": f"{quantity:.4f}", "N52_KAINA": f"{price:.2f}", "N52_OP_DATA": op_date, "N52_BEG_DATE": start_date.isoformat(), } if price_list_code: n52["N52_KODAS_K0"] = price_list_code if end_date is not None: n52["N52_END_DATE"] = end_date.isoformat() payload = { "method": "EDIT_N52", "params": { "oper": "I", "user": api_key.split(".", 1)[0], }, "data": {"N52": n52}, } _post(api_key, payload) def update_contract_n51( contract_code: str, client_code: str, title: str, start_date: date, end_date: Optional[date] = None, comment: Optional[str] = None, status_code: Optional[str] = None, contract_type: Optional[str] = None, ) -> None: api_key = os.getenv("RIVILE_API_KEY", "").strip() if not api_key: raise RuntimeError("Missing RIVILE_API_KEY environment variable.") n51 = { "N51_KODAS_KT": contract_code, "N51_KODAS_KS": client_code, "N51_PAV": title, "N51_OP_DATA": start_date.isoformat(), "N51_BEG_DATE": start_date.isoformat(), "N51_PASTABOS": comment or "", } if end_date: n51["N51_END_DATE"] = end_date.isoformat() if status_code is not None: n51["N51_VISKAS"] = status_code if contract_type: n51["N51_TIPAS"] = contract_type payload = { "method": "EDIT_N51", "params": { "oper": "U", "user": api_key.split(".", 1)[0], "fld": "N51_KODAS_KT", "val": contract_code, }, "data": {"N51": n51}, } _post(api_key, payload) def update_contract_appendix_n52( contract_code: str, price_list_code: str, service_code: str, price: float, start_date: date, end_date: Optional[date] = None, quantity: float = 1.0, item_type: str = "service", match_service_code: Optional[str] = None, line_no: Optional[int] = None, ) -> None: api_key = os.getenv("RIVILE_API_KEY", "").strip() if not api_key: raise RuntimeError("Missing RIVILE_API_KEY environment variable.") rusis = "2" if item_type == "service" else "1" op_date = date.today().isoformat() n52 = { "N52_KODAS_KT": contract_code, "N52_KODAS_K0": price_list_code, "N52_RUSIS": rusis, "N52_KODAS": service_code, "N52_KIEKIS": f"{quantity:.4f}", "N52_KAINA": f"{price:.2f}", "N52_OP_DATA": op_date, "N52_BEG_DATE": start_date.isoformat(), } if end_date is not None: n52["N52_END_DATE"] = end_date.isoformat() else: n52["N52_END_DATE"] = "" key_service_code = (match_service_code or service_code).strip() if line_no is not None: n52["N52_EIL_NR"] = line_no fld = "N52_KODAS_KT,N52_KODAS_K0,N52_KODAS,N52_EIL_NR" val = f"{contract_code},{price_list_code},{key_service_code},{line_no}" else: fld = "N52_KODAS_KT,N52_KODAS_K0,N52_KODAS" val = f"{contract_code},{price_list_code},{key_service_code}" payload = { "method": "EDIT_N52", "params": { "oper": "U", "user": api_key.split(".", 1)[0], "fld": fld, "val": val, }, "data": {"N52": n52}, } _post(api_key, payload) def delete_contract_appendix_n52( contract_code: str, price_list_code: str, service_code: Optional[str] = None, ) -> None: api_key = os.getenv("RIVILE_API_KEY", "").strip() if not api_key: raise RuntimeError("Missing RIVILE_API_KEY environment variable.") n52 = { "N52_KODAS_KT": contract_code, "N52_KODAS_K0": price_list_code, } if service_code: n52["N52_KODAS"] = service_code if service_code: fld = "N52_KODAS_KT,N52_KODAS_K0,N52_KODAS" val = f"{contract_code},{price_list_code},{service_code}" else: fld = "N52_KODAS_KT,N52_KODAS_K0" val = f"{contract_code},{price_list_code}" payload = { "method": "EDIT_N52", "params": { "oper": "D", "user": api_key.split(".", 1)[0], "fld": fld, "val": val, }, "data": {"N52": n52}, } _post(api_key, payload) def _generate_invoice_doc_number(invoice_date: date) -> str: return f"SF-{invoice_date.strftime('%Y%m%d')}-{uuid.uuid4().hex[:4].upper()}" def create_invoice_i06( client_code: str, invoice_date: date, document_number: Optional[str] = None, vat_code: str = "PVM", currency: str = "EUR", comment: Optional[str] = None, ) -> str: api_key = os.getenv("RIVILE_API_KEY", "").strip() if not api_key: raise RuntimeError("Missing RIVILE_API_KEY environment variable.") doc_number = document_number or _generate_invoice_doc_number(invoice_date) i06 = { "I06_OP_TIP": "51", "I06_DOK_NR": doc_number, "I06_OP_DATA": invoice_date.isoformat(), "I06_DOK_DATA": invoice_date.isoformat(), "I06_KODAS_KS": client_code, "I06_KODAS_XS": vat_code, "I06_KODAS_VL": currency, "I06_PASTABOS": comment or "", } payload = { "method": "EDIT_I06", "params": {"oper": "I"}, "data": {"I06": i06}, } resp = _post(api_key, payload) return resp["I06"]["I06_KODAS_PO"] def create_invoice_line_i07( invoice_code: str, item_code: str, quantity: float, unit_price: float, unit_code: Optional[str] = None, line_no: int = 1, division_code: str = "01", item_type: str = "item", ) -> None: api_key = os.getenv("RIVILE_API_KEY", "").strip() if not api_key: raise RuntimeError("Missing RIVILE_API_KEY environment variable.") i07 = { "I07_KODAS_PO": invoice_code, "I07_EIL_NR": line_no, "I07_KODAS": item_code, "I07_KODAS_IS": division_code, "I07_KIEKIS": quantity, "I07_KAINA": f"{unit_price:.2f}", } if item_type == "service": i07["I07_KODAS_PS"] = item_code if unit_code: i07["I07_KODAS_US"] = unit_code i07["I07_KODAS_US_P"] = unit_code i07["I07_KODAS_US_A"] = unit_code payload = { "method": "EDIT_I07", "params": {"oper": "I"}, "data": {"I07": i07}, } _post(api_key, payload) def create_invoice_with_line( client_code: str, item_code: str, quantity: float, unit_price: float, invoice_date: date, comment: Optional[str] = None, item_type: str = "item", unit_code: Optional[str] = None, ) -> Dict[str, str]: invoice_code = create_invoice_i06( client_code=client_code, invoice_date=invoice_date, comment=comment, ) create_invoice_line_i07( invoice_code=invoice_code, item_code=item_code, quantity=quantity, unit_price=unit_price, item_type=item_type, unit_code=unit_code, ) return {"invoice_code": invoice_code}