from __future__ import annotations import argparse 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 _parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument("--user-id", dest="user_id") parser.add_argument("--sutartis-id", dest="sutartis_id") return parser.parse_args() def _get_sutarties_kodas(args: argparse.Namespace) -> str: sutarties_kodas = (args.sutartis_id or "").strip() if sutarties_kodas: return sutarties_kodas return input("Sutarties kodas: ").strip() def _get_user_id(args: argparse.Namespace) -> str: return (args.user_id or "").strip() def main() -> None: args = _parse_args() sutarties_kodas = _get_sutarties_kodas(args) 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 = _get_user_id(args) or 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()