from __future__ import annotations import os from datetime import date, datetime from pathlib import Path from typing import Dict, Iterable, Optional import psycopg2 import psycopg2.extras import requests from dotenv import load_dotenv from uv_app.core.mssql import connect_to_mssql from uv_app.core.pgsql import connect_to_pgsql BASE_URL = "https://api.manorivile.lt/client/v2" TIMEOUT = 30 MIN_YEAR = 2000 DOTENV_PATH = Path(__file__).resolve().parents[2] / ".env" load_dotenv(DOTENV_PATH, override=True) QUERY_PATH = Path(__file__).with_name("sutartys_by_sutarties_kodas.sql") 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_api_key() -> str: api_key = os.getenv("RIVILE_API_KEY", "").strip() if not api_key: raise RuntimeError("Missing RIVILE_API_KEY environment variable.") return api_key def _read_query() -> str: return QUERY_PATH.read_text(encoding="utf-8") def _fetch_rows(sutarties_kodas: str) -> Iterable[Dict[str, object]]: conn = connect_to_pgsql() if conn is None: raise RuntimeError("Failed to connect to PostgreSQL.") try: with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cursor: cursor.execute(_read_query(), (sutarties_kodas,)) return cursor.fetchall() finally: conn.close() def _contract_exists(conn: "pyodbc.Connection", contract_code: str) -> bool: cursor = conn.cursor() cursor.execute( """ SELECT 1 FROM dbo.N51_SUTH WHERE N51_KODAS_KT = ? """, (contract_code,), ) return cursor.fetchone() is not None def _fetch_existing_contract( conn: "pyodbc.Connection", contract_code: str, ) -> Optional[Dict[str, object]]: cursor = conn.cursor() cursor.execute( """ SELECT N51_KODAS_KT, N51_KODAS_KS, N51_PAV, CONVERT(date, N51_OP_DATA) AS N51_OP_DATA, CONVERT(date, N51_BEG_DATE) AS N51_BEG_DATE, N51_TIPAS, N51_APRASYMAS1, N51_APRASYMAS2 FROM dbo.N51_SUTH WHERE N51_KODAS_KT = ? """, (contract_code,), ) row = cursor.fetchone() if not row: return None columns = [c[0] for c in cursor.description] return dict(zip(columns, row)) def _normalize_text(value: object) -> str: if value is None: return "" return str(value).strip() def _normalize_date(value: object) -> date: if isinstance(value, date): result = value elif isinstance(value, datetime): result = value.date() else: return date.today() if result < date(MIN_YEAR, 1, 1): return date.today() return result def _is_desired_contract( existing: Optional[Dict[str, object]], contract_code: str, client_code: str, contract_type: str, start: date, address: str, password: str, ) -> bool: if not existing: return False return ( _normalize_text(existing.get("N51_KODAS_KT")) == contract_code and _normalize_text(existing.get("N51_KODAS_KS")) == client_code and _normalize_text(existing.get("N51_PAV")) == contract_code and _normalize_date(existing.get("N51_OP_DATA")) == start and _normalize_date(existing.get("N51_BEG_DATE")) == start and _normalize_text(existing.get("N51_TIPAS")) == contract_type and _normalize_text(existing.get("N51_APRASYMAS1")) == address and _normalize_text(existing.get("N51_APRASYMAS2")) == password ) def _choose_contract_type(conn: "pyodbc.Connection") -> Optional[str]: cursor = conn.cursor() cursor.execute( """ SELECT N51_TIPAS, COUNT(*) AS cnt FROM dbo.N51_SUTH WHERE N51_TIPAS IS NOT NULL GROUP BY N51_TIPAS ORDER BY cnt DESC, N51_TIPAS """ ) row = cursor.fetchone() if not row: return None return str(row[0]).strip() def _build_address(row: Dict[str, object]) -> str: parts = [] gatve = (row.get("gatve") or "").strip() namas = (row.get("namas") or "").strip() miestas = (row.get("miestas") or "").strip() rajonas = (row.get("rajonas") or "").strip() if gatve or namas: parts.append(" ".join(p for p in [gatve, namas] if p)) if miestas: parts.append(miestas) if rajonas: parts.append(rajonas) return ", ".join(parts) def _create_contract( api_key: str, contract_code: str, client_code: str, start: date, contract_type: str, address: str, password: str, ) -> None: payload = { "method": "EDIT_N51_FULL", "params": { "oper": "I", "user": api_key.split(".", 1)[0], }, "data": { "N51": { "N51_KODAS_KT": contract_code, "N51_KODAS_KS": client_code, "N51_PAV": contract_code, "N51_OP_DATA": start.isoformat(), "N51_BEG_DATE": start.isoformat(), "N51_TIPAS": contract_type, "N51_APRASYMAS1": address, "N51_APRASYMAS2": password, } }, } _post(api_key, payload) def _update_contract( api_key: str, contract_code: str, client_code: str, contract_type: str, address: str, password: str, start: date, ) -> None: payload = { "method": "EDIT_N51", "params": { "oper": "U", "user": api_key.split(".", 1)[0], "fld": "N51_KODAS_KT", "val": contract_code, }, "data": { "N51": { "N51_KODAS_KT": contract_code, "N51_KODAS_KS": client_code, "N51_PAV": contract_code, "N51_TIPAS": contract_type, "N51_OP_DATA": start.isoformat(), "N51_BEG_DATE": start.isoformat(), "N51_APRASYMAS1": address, "N51_APRASYMAS2": password, } }, } _post(api_key, payload) def main() -> None: sutarties_kodas = input("Sutarties kodas: ").strip() if not sutarties_kodas: print("Missing sutarties kodas.") return rows = list(_fetch_rows(sutarties_kodas)) if not rows: print("No sutartis rows found.") return row = rows[0] client_code = str(row.get("client_code") or "").strip() if not client_code: raise RuntimeError("Missing client code for sutartis.") contract_code = f"SUT-{sutarties_kodas}" start_date = _normalize_date(row.get("data")) address = _build_address(row) password = str(row.get("password") or "").strip() api_key = _get_api_key() mssql_conn = connect_to_mssql() if mssql_conn is None: raise RuntimeError("Failed to connect to MSSQL.") try: contract_type = _choose_contract_type(mssql_conn) if not contract_type: raise RuntimeError("Failed to resolve N51_TIPAS for Pardavimo.") existing = _fetch_existing_contract(mssql_conn, contract_code) if _is_desired_contract( existing, contract_code, client_code, contract_type, start_date, address, password, ): print("Already in desired state; no changes.") return if existing: _update_contract( api_key, contract_code, client_code, contract_type, address, password, start_date, ) print(f"Updated contract {contract_code}") else: _create_contract( api_key, contract_code, client_code, start_date, contract_type, address, password, ) print(f"Created contract {contract_code}") finally: mssql_conn.close() if __name__ == "__main__": main()