""" Rajvidya Astro API — thin Python SDK. Usage example: from rajvidya import RajvidyaClient, RajvidyaError client = RajvidyaClient(api_key='rv_prod_ABC_xyzxyzxyz') # GET endpoint with parameters try: kundli = client.kundli(chart_id='chart-123', lang='en') print(f"Credits charged: {kundli.credits_charged}") print(f"Moon sign: {kundli.data['moon_sign']}") except RajvidyaError as e: print(f"API error: {e}") # POST endpoint chart = client.create_chart( name='John Doe', birth_date='1990-05-15', birth_time='14:30:00', birth_place='New York, USA', latitude=40.7128, longitude=-74.0060, ) print(f"Created chart: {chart.data['id']}") """ import json import urllib.request import urllib.error from typing import Any, Dict, Optional class RajvidyaError(Exception): """Raised when the API returns a non-2xx status code.""" def __init__(self, status_code: int, code: str, message: str): self.status_code = status_code self.code = code self.message = message super().__init__(f"{code}: {message}") class ApiResponse: """Wraps a successful API response with metadata.""" def __init__(self, status_code: int, data: Dict[str, Any], headers: Dict[str, str]): self.status_code = status_code self.data = data self.headers = headers self.credits_charged = headers.get('X-Credits-Charged') self.credits_remaining = headers.get('X-Credits-Remaining') self.quota_reset_at = headers.get('X-Quota-Reset-At') class RajvidyaClient: """Thin HTTP client for the Rajvidya Astro API.""" def __init__(self, api_key: str, base_url: str = 'https://rajvidya.com'): """ Args: api_key: Your API key (format: rv_prod_ABC_xyzxyzxyz) base_url: Base URL of the API server """ self.api_key = api_key self.base_url = base_url.rstrip('/') def _request(self, method: str, path: str, params: Optional[Dict[str, Any]] = None, json_body: Optional[Dict[str, Any]] = None) -> ApiResponse: """Execute an HTTP request and return the parsed response.""" url = f"{self.base_url}{path}" if params: query_string = '&'.join( f"{k}={v}" for k, v in params.items() if v is not None ) if query_string: url = f"{url}?{query_string}" headers = { 'X-API-Key': self.api_key, 'User-Agent': 'rajvidya-python-sdk/1.0', } body = None if json_body is not None: body = json.dumps(json_body).encode('utf-8') headers['Content-Type'] = 'application/json' req = urllib.request.Request(url, data=body, headers=headers, method=method) try: with urllib.request.urlopen(req) as response: response_data = json.loads(response.read().decode('utf-8')) response_headers = dict(response.headers) return ApiResponse(response.status, response_data, response_headers) except urllib.error.HTTPError as e: try: error_data = json.loads(e.read().decode('utf-8')) error_obj = error_data.get('error', {}) code = error_obj.get('code', 'unknown_error') message = error_obj.get('message', str(e)) except (json.JSONDecodeError, KeyError): code = 'http_error' message = str(e) raise RajvidyaError(e.code, code, message) except urllib.error.URLError as e: raise RajvidyaError(0, 'network_error', f"Network error: {e.reason}") # Chart management def create_chart(self, name: str, birth_date: str, birth_time: str, birth_place: str, latitude: float, longitude: float, gender: str = 'MALE', timezone_offset: Optional[float] = None) -> ApiResponse: """POST /api/v1/charts/ — Create a birth record.""" return self._request('POST', '/api/v1/charts/', json_body={ 'name': name, 'gender': gender, 'date_of_birth': birth_date, 'time_of_birth': birth_time, 'place_of_birth': birth_place, 'latitude': latitude, 'longitude': longitude, 'timezone_offset': timezone_offset, }) def read_chart(self, chart_id: str) -> ApiResponse: """GET /api/v1/charts/ — Read a birth record.""" return self._request('GET', '/api/v1/charts/', params={'chart_id': chart_id}) def update_chart(self, chart_id: str, **updates) -> ApiResponse: """POST /api/v1/charts/ — Update a birth record.""" updates['chart_id'] = chart_id return self._request('POST', '/api/v1/charts/', json_body=updates) # Kundli & birth chart def kundli(self, chart_id: str, lang: str = 'en') -> ApiResponse: """GET /api/v1/kundli/ — Full birth chart with planets, houses, yogas.""" return self._request('GET', '/api/v1/kundli/', params={'chart_id': chart_id, 'lang': lang}) # Dasha def dasha(self, chart_id: str, lang: str = 'en') -> ApiResponse: """GET /api/v1/dasha/ — Dasha timeline from birth to 120 years.""" return self._request('GET', '/api/v1/dasha/', params={'chart_id': chart_id, 'lang': lang}) # Doshas def doshas(self, chart_id: str, lang: str = 'en') -> ApiResponse: """GET /api/v1/doshas/ — Manglik, Kaal Sarp, Pitra, Sade Sati with severity.""" return self._request('GET', '/api/v1/doshas/', params={'chart_id': chart_id, 'lang': lang}) # Horoscope def rashifal(self, chart_id: str, lang: str = 'en') -> ApiResponse: """GET /api/v1/rashifal/ — Moon-sign horoscope from live transits.""" return self._request('GET', '/api/v1/rashifal/', params={'chart_id': chart_id, 'lang': lang}) def today_horoscope(self, chart_id: str, lang: str = 'en') -> ApiResponse: """GET /api/v1/today/ — Today's horoscope by transit.""" return self._request('GET', '/api/v1/today/', params={'chart_id': chart_id, 'lang': lang}) # Transits & live sky def live_sky(self, latitude: float, longitude: float, lang: str = 'en') -> ApiResponse: """GET /api/v1/live-sky/ — Real-time planetary positions.""" return self._request('GET', '/api/v1/live-sky/', params={ 'latitude': latitude, 'longitude': longitude, 'lang': lang, }) def gochar(self, chart_id: str, lang: str = 'en') -> ApiResponse: """GET /api/v1/gochar/ — Transit effects on the natal chart.""" return self._request('GET', '/api/v1/gochar/', params={'chart_id': chart_id, 'lang': lang}) # Panchang def panchang(self, latitude: float, longitude: float, lang: str = 'en') -> ApiResponse: """GET /api/v1/panchang/ — Tithi, nakshatra, choghadiya for a location.""" return self._request('GET', '/api/v1/panchang/', params={ 'latitude': latitude, 'longitude': longitude, 'lang': lang, }) def festivals(self, latitude: float, longitude: float, lang: str = 'en') -> ApiResponse: """GET /api/v1/festivals/ — Festival dates for a location.""" return self._request('GET', '/api/v1/festivals/', params={ 'latitude': latitude, 'longitude': longitude, 'lang': lang, }) def choghadiya(self, latitude: float, longitude: float, lang: str = 'en') -> ApiResponse: """GET /api/v1/choghadiya/ — Auspicious time windows for a location.""" return self._request('GET', '/api/v1/choghadiya/', params={ 'latitude': latitude, 'longitude': longitude, 'lang': lang, }) # Muhurta def muhurta_marriage(self, bride_chart_id: str, groom_chart_id: str, lang: str = 'en') -> ApiResponse: """GET /api/v1/muhurta/marriage/ — Auspicious marriage dates.""" return self._request('GET', '/api/v1/muhurta/marriage/', params={ 'bride_chart_id': bride_chart_id, 'groom_chart_id': groom_chart_id, 'lang': lang, }) # Compatibility def compatibility(self, chart1_id: str, chart2_id: str, lang: str = 'en') -> ApiResponse: """GET /api/v1/compatibility/ — Twenty-plus compatibility modules.""" return self._request('GET', '/api/v1/compatibility/', params={ 'chart1_id': chart1_id, 'chart2_id': chart2_id, 'lang': lang, }) def matching_report(self, bride_chart_id: str, groom_chart_id: str, lang: str = 'en') -> ApiResponse: """GET /api/v1/reports/matching/ — Full matrimony analysis.""" return self._request('GET', '/api/v1/reports/matching/', params={ 'bride_chart_id': bride_chart_id, 'groom_chart_id': groom_chart_id, 'lang': lang, }) # Numerology def numerology_dashboard(self, name: str, lang: str = 'en') -> ApiResponse: """GET /api/v1/numerology/dashboard/ — Chaldean and Vedic numbers.""" return self._request('GET', '/api/v1/numerology/dashboard/', params={ 'name': name, 'lang': lang, }) def numerology_pro(self, name: str, lang: str = 'en') -> ApiResponse: """GET /api/v1/numerology/pro/ — Advanced numerology analysis.""" return self._request('GET', '/api/v1/numerology/pro/', params={ 'name': name, 'lang': lang, }) def baby_names(self, birth_chart_id: str, lang: str = 'en') -> ApiResponse: """GET /api/v1/baby-names/ — Name suggestions for a birth time.""" return self._request('GET', '/api/v1/baby-names/', params={ 'birth_chart_id': birth_chart_id, 'lang': lang, }) # Biorhythm def biorhythm(self, birth_date: str, lang: str = 'en') -> ApiResponse: """GET /api/v1/biorhythm/ — 23/28/33 day biorhythm cycles.""" return self._request('GET', '/api/v1/biorhythm/', params={ 'birth_date': birth_date, 'lang': lang, }) # Remedies def remedies_prescription(self, chart_id: str, lang: str = 'en') -> ApiResponse: """GET /api/v1/remedies/prescription/ — Gemstone, mantra, charity prescriptions.""" return self._request('GET', '/api/v1/remedies/prescription/', params={ 'chart_id': chart_id, 'lang': lang, }) # Vastu def vastu_analyze(self, floor_plan_base64: str, building_type: str, orientation_degrees: int, lang: str = 'en') -> ApiResponse: """POST /api/v1/vastu/analyze/ — Analyze a floor plan.""" return self._request('POST', '/api/v1/vastu/analyze/', json_body={ 'floor_plan_base64': floor_plan_base64, 'building_type': building_type, 'orientation_degrees': orientation_degrees, 'lang': lang, }) def vastu_surroundings(self, latitude: float, longitude: float, lang: str = 'en') -> ApiResponse: """GET /api/v1/vastu/surroundings/ — Vastu score for surroundings.""" return self._request('GET', '/api/v1/vastu/surroundings/', params={ 'latitude': latitude, 'longitude': longitude, 'lang': lang, }) def vastu_catalog(self, lang: str = 'en') -> ApiResponse: """GET /api/v1/vastu/catalog/ — Vastu defect catalog.""" return self._request('GET', '/api/v1/vastu/catalog/', params={'lang': lang})